From ca78f73e0a9e306cec9ff444f9a3c4efb925397d Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Fri, 25 Oct 2024 19:41:10 +0200 Subject: [PATCH] Introduce Log Persistence Strategy (#6057) * Implement log persistence strategy management Added interfaces, services, and strategies for log persistence. Introduced new endpoint to list available log persistence strategies. Updated configurations and dependency injections accordingly. * Refactor log record methods to asynchronous Updated methods for extracting and persisting log records to be asynchronous, enhancing performance and scalability. This change includes modifying interfaces and implementations for better async support in workflow execution logging. * Remove commented code * Support nullable values in ActivityState dictionaries Update ActivityState to support nullable values by changing type to 'IDictionary'. Enhanced DefaultActivityExecutionMapper to handle multiple persistence strategies for logging inputs and outputs. * Rename ShouldPersistAsync to GetPersistenceModeAsync Refactor method names for log persistence strategies to improve readability and consistency. Added summary comments for clarification and removed redundant configurations from appsettings.json. Added implicit uses and updated namespaces for better maintainability. * Refactor activity payload and output retrieval logic Extract payload and output retrieval into `GetPayload` and `GetOutputs` methods respectively. This modularizes the code for better readability and maintainability, and allows for potential reusability of these methods in other parts of the codebase. * Add new project reference and update PostgreSQL provider usage Added a project reference to Elsa.Agents.Persistence.EntityFrameworkCore.PostgreSql in the test project file. Also modified the WorkflowServer setup to specify the assembly in the PostgreSQL provider configuration. * Add agent persistence to WorkflowServer Integrated agent support and persistence using PostgreSQL in WorkflowServer. This includes adding necessary project references and configuring agents in the workflow server setup. --- src/apps/Elsa.Server.Web/Program.cs | 3 +- src/apps/Elsa.Server.Web/appsettings.json | 9 +- .../Elsa.Api.Client.csproj.DotSettings | 4 +- .../DependencyInjectionExtensions.cs | 2 + .../Contracts/ILogPersistenceStrategiesApi.cs | 19 ++ .../LogPersistenceStrategyDescriptor.cs | 12 ++ .../LogPersistenceStrategies/List/Endpoint.cs | 45 +++++ .../Contexts/ActivityExecutionContext.cs | 2 +- .../Elsa.Workflows.Core.csproj.DotSettings | 5 + .../Features/WorkflowsFeature.cs | 11 +- .../Contexts/LogPersistenceStrategyContext.cs | 3 + .../Contracts/ILogPersistenceStrategy.cs | 6 + .../ILogPersistenceStrategyService.cs | 6 + .../Enums/LogPersistenceMode.cs | 48 ++--- .../Options/LogPersistenceOptions.cs | 6 + .../DefaultLogPersistenceStrategyService.cs | 12 ++ .../Strategies/Configuration.cs | 16 ++ .../LogPersistence/Strategies/Exclude.cs | 15 ++ .../LogPersistence/Strategies/Include.cs | 15 ++ .../LogPersistence/Strategies/Inherit.cs | 15 ++ .../Features/WorkflowManagementFeature.cs | 1 + .../Options/ManagementOptions.cs | 1 + .../Activities/BulkDispatchWorkflows.cs | 1 - .../Contracts/IActivityExecutonMapper.cs | 2 +- .../Contracts/ILogRecordExtractor.cs | 2 +- .../Entities/ActivityExecutionRecord.cs | 2 +- .../ActivityExecutionRecordExtractor.cs | 5 +- .../DefaultActivityExecutionMapper.cs | 170 +++++++++++++----- .../Services/StoreActivityExecutionLogSink.cs | 3 +- .../Services/StoreWorkflowExecutionLogSink.cs | 3 +- .../WorkflowExecutionLogRecordExtractor.cs | 6 +- ...Elsa.AzureServiceBus.ComponentTests.csproj | 1 + .../Fixtures/WorkflowServer.cs | 3 + .../Elsa.Workflows.ComponentTests.csproj | 1 + .../Helpers/Fixtures/WorkflowServer.cs | 3 +- 35 files changed, 363 insertions(+), 95 deletions(-) create mode 100644 src/clients/Elsa.Api.Client/Resources/LogPersistenceStrategies/Contracts/ILogPersistenceStrategiesApi.cs create mode 100644 src/clients/Elsa.Api.Client/Resources/LogPersistenceStrategies/Models/LogPersistenceStrategyDescriptor.cs create mode 100644 src/modules/Elsa.Workflows.Api/Endpoints/LogPersistenceStrategies/List/Endpoint.cs create mode 100644 src/modules/Elsa.Workflows.Core/LogPersistence/Contexts/LogPersistenceStrategyContext.cs create mode 100644 src/modules/Elsa.Workflows.Core/LogPersistence/Contracts/ILogPersistenceStrategy.cs create mode 100644 src/modules/Elsa.Workflows.Core/LogPersistence/Contracts/ILogPersistenceStrategyService.cs rename src/modules/Elsa.Workflows.Core/{ => LogPersistence}/Enums/LogPersistenceMode.cs (89%) create mode 100644 src/modules/Elsa.Workflows.Core/LogPersistence/Options/LogPersistenceOptions.cs create mode 100644 src/modules/Elsa.Workflows.Core/LogPersistence/Services/DefaultLogPersistenceStrategyService.cs create mode 100644 src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Configuration.cs create mode 100644 src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Exclude.cs create mode 100644 src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Include.cs create mode 100644 src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Inherit.cs diff --git a/src/apps/Elsa.Server.Web/Program.cs b/src/apps/Elsa.Server.Web/Program.cs index 22620706f..65f3d6063 100644 --- a/src/apps/Elsa.Server.Web/Program.cs +++ b/src/apps/Elsa.Server.Web/Program.cs @@ -3,7 +3,6 @@ using Elsa.Agents; using Elsa.Alterations.Extensions; using Elsa.Alterations.MassTransit.Extensions; using Elsa.Common.DistributedLocks.Noop; -using Elsa.Common.Multitenancy; using Elsa.Dapper.Extensions; using Elsa.Dapper.Services; using Elsa.DropIns.Extensions; @@ -29,8 +28,8 @@ using Elsa.Server.Web.Extensions; using Elsa.Server.Web.Filters; using Elsa.Tenants.AspNetCore; using Elsa.Tenants.Extensions; -using Elsa.Workflows; using Elsa.Workflows.Api; +using Elsa.Workflows.LogPersistence; using Elsa.Workflows.Management.Compression; using Elsa.Workflows.Management.Stores; using Elsa.Workflows.Runtime.Distributed.Extensions; diff --git a/src/apps/Elsa.Server.Web/appsettings.json b/src/apps/Elsa.Server.Web/appsettings.json index fb46c79e8..a47727e5a 100644 --- a/src/apps/Elsa.Server.Web/appsettings.json +++ b/src/apps/Elsa.Server.Web/appsettings.json @@ -2,14 +2,7 @@ "Logging": { "LogLevel": { "Default": "Warning", - "Elsa": "Warning", - "MassTransit": "Warning", - "Microsoft.Extensions.Http": "Warning", - "Microsoft.Hosting.Lifetime": "Information", - "Microsoft.EntityFrameworkCore": "Warning", - "Microsoft.AspNetCore": "Warning", - "Quartz": "Warning", - "System.Net.Http": "Warning" + "Microsoft.Hosting.Lifetime": "Information" } }, "AllowedHosts": "*", diff --git a/src/clients/Elsa.Api.Client/Elsa.Api.Client.csproj.DotSettings b/src/clients/Elsa.Api.Client/Elsa.Api.Client.csproj.DotSettings index 8b01856ae..801d2b040 100644 --- a/src/clients/Elsa.Api.Client/Elsa.Api.Client.csproj.DotSettings +++ b/src/clients/Elsa.Api.Client/Elsa.Api.Client.csproj.DotSettings @@ -1,2 +1,4 @@  - True \ No newline at end of file + True + True + True \ No newline at end of file diff --git a/src/clients/Elsa.Api.Client/Extensions/DependencyInjectionExtensions.cs b/src/clients/Elsa.Api.Client/Extensions/DependencyInjectionExtensions.cs index 9f28a6070..4be3d8479 100644 --- a/src/clients/Elsa.Api.Client/Extensions/DependencyInjectionExtensions.cs +++ b/src/clients/Elsa.Api.Client/Extensions/DependencyInjectionExtensions.cs @@ -5,6 +5,7 @@ using Elsa.Api.Client.Resources.ActivityExecutions.Contracts; using Elsa.Api.Client.Resources.Features.Contracts; using Elsa.Api.Client.Resources.Identity.Contracts; using Elsa.Api.Client.Resources.IncidentStrategies.Contracts; +using Elsa.Api.Client.Resources.LogPersistenceStrategies; using Elsa.Api.Client.Resources.Scripting.Contracts; using Elsa.Api.Client.Resources.StorageDrivers.Contracts; using Elsa.Api.Client.Resources.Tasks.Contracts; @@ -65,6 +66,7 @@ public static class DependencyInjectionExtensions services.AddApi(builderOptions); services.AddApi(builderOptions); services.AddApi(builderOptions); + services.AddApi(builderOptions); services.AddApi(builderOptions); services.AddApi(builderOptions); services.AddApi(builderOptions); diff --git a/src/clients/Elsa.Api.Client/Resources/LogPersistenceStrategies/Contracts/ILogPersistenceStrategiesApi.cs b/src/clients/Elsa.Api.Client/Resources/LogPersistenceStrategies/Contracts/ILogPersistenceStrategiesApi.cs new file mode 100644 index 000000000..94f0fed61 --- /dev/null +++ b/src/clients/Elsa.Api.Client/Resources/LogPersistenceStrategies/Contracts/ILogPersistenceStrategiesApi.cs @@ -0,0 +1,19 @@ +using Elsa.Api.Client.Shared.Models; +using JetBrains.Annotations; +using Refit; + +namespace Elsa.Api.Client.Resources.LogPersistenceStrategies; + +/// +/// Represents a client API for managing log persistence strategies. +/// +[UsedImplicitly] +public interface ILogPersistenceStrategiesApi +{ + /// + /// Lists log persistence strategies. + /// + /// The cancellation token. + [Get("/descriptors/log-persistence-strategies")] + Task> ListAsync(CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/clients/Elsa.Api.Client/Resources/LogPersistenceStrategies/Models/LogPersistenceStrategyDescriptor.cs b/src/clients/Elsa.Api.Client/Resources/LogPersistenceStrategies/Models/LogPersistenceStrategyDescriptor.cs new file mode 100644 index 000000000..89cccfc46 --- /dev/null +++ b/src/clients/Elsa.Api.Client/Resources/LogPersistenceStrategies/Models/LogPersistenceStrategyDescriptor.cs @@ -0,0 +1,12 @@ +using JetBrains.Annotations; + +namespace Elsa.Api.Client.Resources.LogPersistenceStrategies; + +/// +/// Represents a log persistence strategy. +/// +/// The .NET type name of the log persistence strategy. +/// The display name of the log persistence strategy. +/// The description of the log persistence strategy. +[UsedImplicitly] +public record LogPersistenceStrategyDescriptor(string TypeName, string DisplayName, string? Description); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/LogPersistenceStrategies/List/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/LogPersistenceStrategies/List/Endpoint.cs new file mode 100644 index 000000000..0dd4992c0 --- /dev/null +++ b/src/modules/Elsa.Workflows.Api/Endpoints/LogPersistenceStrategies/List/Endpoint.cs @@ -0,0 +1,45 @@ +using System.ComponentModel; +using System.ComponentModel.DataAnnotations; +using System.Reflection; +using Elsa.Abstractions; +using Elsa.Extensions; +using Elsa.Models; +using Elsa.Workflows.LogPersistence; +using Humanizer; + +namespace Elsa.Workflows.Api.Endpoints.LogPersistenceStrategies.List; + +/// +/// Returns list of available implementations. +/// +internal class Endpoint(ILogPersistenceStrategyService logPersistenceStrategyService) : ElsaEndpointWithoutRequest> +{ + public override void Configure() + { + Get("/descriptors/log-persistence-strategies"); + ConfigurePermissions("read:log-persistence-strategies"); + } + + public override Task> ExecuteAsync(CancellationToken cancellationToken) + { + var strategies = logPersistenceStrategyService.ListStrategies(); + var descriptors = strategies.Select(LogPersistenceStrategyDescriptor.FromStrategy).OrderBy(x => x.DisplayName).ToList(); + var response =new ListResponse(descriptors); + return Task.FromResult(response); + } +} + +internal record LogPersistenceStrategyDescriptor(string DisplayName, string Description, string TypeName) +{ + public static LogPersistenceStrategyDescriptor FromStrategy(ILogPersistenceStrategy strategy) + { + var type = strategy.GetType(); + var displayNameAttribute = type.GetCustomAttribute(); + var descriptionAttribute = type.GetCustomAttribute(); + var displayAttribute = type.GetCustomAttribute(); + var displayName = displayNameAttribute?.DisplayName ?? displayAttribute?.Name ?? type.Name.Replace("Strategy", "").Humanize(); + var description = descriptionAttribute?.Description ?? displayAttribute?.Description ?? ""; + + return new LogPersistenceStrategyDescriptor(displayName, description, type.GetSimpleAssemblyQualifiedName()); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs index 5a96c696f..16ff509bd 100644 --- a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs +++ b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs @@ -195,7 +195,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable /// /// Stores the evaluated inputs, serialized, for the current activity for historical purposes. /// - public IDictionary ActivityState { get; set; } = new Dictionary(); + public IDictionary ActivityState { get; set; } = new Dictionary(); /// /// Schedules the specified activity to be executed. diff --git a/src/modules/Elsa.Workflows.Core/Elsa.Workflows.Core.csproj.DotSettings b/src/modules/Elsa.Workflows.Core/Elsa.Workflows.Core.csproj.DotSettings index a9f483ae3..57af41798 100644 --- a/src/modules/Elsa.Workflows.Core/Elsa.Workflows.Core.csproj.DotSettings +++ b/src/modules/Elsa.Workflows.Core/Elsa.Workflows.Core.csproj.DotSettings @@ -12,5 +12,10 @@ True True True + True + True + True + True + True True True \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Features/WorkflowsFeature.cs b/src/modules/Elsa.Workflows.Core/Features/WorkflowsFeature.cs index 4d8af941b..a73b24160 100644 --- a/src/modules/Elsa.Workflows.Core/Features/WorkflowsFeature.cs +++ b/src/modules/Elsa.Workflows.Core/Features/WorkflowsFeature.cs @@ -9,6 +9,8 @@ using Elsa.Features.Services; using Elsa.Workflows.ActivationValidators; using Elsa.Workflows.Builders; using Elsa.Workflows.IncidentStrategies; +using Elsa.Workflows.LogPersistence; +using Elsa.Workflows.LogPersistence.Strategies; using Elsa.Workflows.Middleware.Activities; using Elsa.Workflows.Middleware.Workflows; using Elsa.Workflows.Pipelines.ActivityExecution; @@ -194,9 +196,16 @@ public class WorkflowsFeature : FeatureBase .AddScoped() .AddScoped() - // Logger state generators + // Logger state generators. .AddSingleton(WorkflowLoggerStateGenerator) .AddSingleton(ActivityLoggerStateGenerator) + + // Log Persistence Strategies. + .AddScoped() + .AddScoped() + .AddScoped() + .AddScoped() + .AddScoped() // Logging .AddLogging(); diff --git a/src/modules/Elsa.Workflows.Core/LogPersistence/Contexts/LogPersistenceStrategyContext.cs b/src/modules/Elsa.Workflows.Core/LogPersistence/Contexts/LogPersistenceStrategyContext.cs new file mode 100644 index 000000000..052278467 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/LogPersistence/Contexts/LogPersistenceStrategyContext.cs @@ -0,0 +1,3 @@ +namespace Elsa.Workflows.LogPersistence; + +public record LogPersistenceStrategyContext(CancellationToken CancellationToken); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/LogPersistence/Contracts/ILogPersistenceStrategy.cs b/src/modules/Elsa.Workflows.Core/LogPersistence/Contracts/ILogPersistenceStrategy.cs new file mode 100644 index 000000000..63d0c1ef6 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/LogPersistence/Contracts/ILogPersistenceStrategy.cs @@ -0,0 +1,6 @@ +namespace Elsa.Workflows.LogPersistence; + +public interface ILogPersistenceStrategy +{ + Task GetPersistenceModeAsync(LogPersistenceStrategyContext context); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/LogPersistence/Contracts/ILogPersistenceStrategyService.cs b/src/modules/Elsa.Workflows.Core/LogPersistence/Contracts/ILogPersistenceStrategyService.cs new file mode 100644 index 000000000..0606c5663 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/LogPersistence/Contracts/ILogPersistenceStrategyService.cs @@ -0,0 +1,6 @@ +namespace Elsa.Workflows.LogPersistence; + +public interface ILogPersistenceStrategyService +{ + IEnumerable ListStrategies(); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Enums/LogPersistenceMode.cs b/src/modules/Elsa.Workflows.Core/LogPersistence/Enums/LogPersistenceMode.cs similarity index 89% rename from src/modules/Elsa.Workflows.Core/Enums/LogPersistenceMode.cs rename to src/modules/Elsa.Workflows.Core/LogPersistence/Enums/LogPersistenceMode.cs index e2ef16466..57518f7ca 100644 --- a/src/modules/Elsa.Workflows.Core/Enums/LogPersistenceMode.cs +++ b/src/modules/Elsa.Workflows.Core/LogPersistence/Enums/LogPersistenceMode.cs @@ -1,25 +1,25 @@ -using JetBrains.Annotations; - -namespace Elsa.Workflows; - -/// -/// Define the Log Persistence mode to store information -/// -[UsedImplicitly] -public enum LogPersistenceMode -{ - /// - /// Persist using the Parent mode - /// - Inherit, - - /// - /// Include property to store - /// - Include, - - /// - /// Exclude Property to store - /// - Exclude +using JetBrains.Annotations; + +namespace Elsa.Workflows.LogPersistence; + +/// +/// Define the Log Persistence mode to store information +/// +[UsedImplicitly] +public enum LogPersistenceMode +{ + /// + /// Persist using the Parent mode + /// + Inherit, + + /// + /// Include property to store + /// + Include, + + /// + /// Exclude Property to store + /// + Exclude } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/LogPersistence/Options/LogPersistenceOptions.cs b/src/modules/Elsa.Workflows.Core/LogPersistence/Options/LogPersistenceOptions.cs new file mode 100644 index 000000000..fbc9be666 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/LogPersistence/Options/LogPersistenceOptions.cs @@ -0,0 +1,6 @@ +namespace Elsa.Workflows.LogPersistence; + +public class LogPersistenceOptions +{ + public LogPersistenceMode ConfiguredPersistenceMode { get; set; } = LogPersistenceMode.Inherit; +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/LogPersistence/Services/DefaultLogPersistenceStrategyService.cs b/src/modules/Elsa.Workflows.Core/LogPersistence/Services/DefaultLogPersistenceStrategyService.cs new file mode 100644 index 000000000..b7f6d4051 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/LogPersistence/Services/DefaultLogPersistenceStrategyService.cs @@ -0,0 +1,12 @@ +using JetBrains.Annotations; + +namespace Elsa.Workflows.LogPersistence; + +[UsedImplicitly] +public class DefaultLogPersistenceStrategyService(IEnumerable strategies) : ILogPersistenceStrategyService +{ + public IEnumerable ListStrategies() + { + return strategies; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Configuration.cs b/src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Configuration.cs new file mode 100644 index 000000000..e73234cce --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Configuration.cs @@ -0,0 +1,16 @@ +using JetBrains.Annotations; +using Microsoft.Extensions.Options; + +namespace Elsa.Workflows.LogPersistence.Strategies; + +/// +/// A log persistence strategy that uses the configured persistence mode. +/// +[UsedImplicitly] +public class Configuration(IOptions options) : ILogPersistenceStrategy +{ + public Task GetPersistenceModeAsync(LogPersistenceStrategyContext context) + { + return Task.FromResult(options.Value.ConfiguredPersistenceMode); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Exclude.cs b/src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Exclude.cs new file mode 100644 index 000000000..f8ace9acd --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Exclude.cs @@ -0,0 +1,15 @@ +using JetBrains.Annotations; + +namespace Elsa.Workflows.LogPersistence.Strategies; + +/// +/// A log persistence strategy that returns . +/// +[UsedImplicitly] +public class Exclude : ILogPersistenceStrategy +{ + public Task GetPersistenceModeAsync(LogPersistenceStrategyContext context) + { + return Task.FromResult(LogPersistenceMode.Exclude); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Include.cs b/src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Include.cs new file mode 100644 index 000000000..7172d900c --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Include.cs @@ -0,0 +1,15 @@ +using JetBrains.Annotations; + +namespace Elsa.Workflows.LogPersistence.Strategies; + +/// +/// A log persistence strategy that returns . +/// +[UsedImplicitly] +public class Include : ILogPersistenceStrategy +{ + public Task GetPersistenceModeAsync(LogPersistenceStrategyContext context) + { + return Task.FromResult(LogPersistenceMode.Include); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Inherit.cs b/src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Inherit.cs new file mode 100644 index 000000000..f2d734a07 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Inherit.cs @@ -0,0 +1,15 @@ +using JetBrains.Annotations; + +namespace Elsa.Workflows.LogPersistence.Strategies; + +/// +/// A log persistence strategy that returns . +/// +[UsedImplicitly] +public class Inherit : ILogPersistenceStrategy +{ + public Task GetPersistenceModeAsync(LogPersistenceStrategyContext context) + { + return Task.FromResult(LogPersistenceMode.Inherit); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Management/Features/WorkflowManagementFeature.cs b/src/modules/Elsa.Workflows.Management/Features/WorkflowManagementFeature.cs index 0f93919cf..d36463f52 100644 --- a/src/modules/Elsa.Workflows.Management/Features/WorkflowManagementFeature.cs +++ b/src/modules/Elsa.Workflows.Management/Features/WorkflowManagementFeature.cs @@ -12,6 +12,7 @@ using Elsa.Features.Abstractions; using Elsa.Features.Attributes; using Elsa.Features.Services; using Elsa.Workflows.Features; +using Elsa.Workflows.LogPersistence; using Elsa.Workflows.Management.Activities.WorkflowDefinitionActivity; using Elsa.Workflows.Management.Compression; using Elsa.Workflows.Management.Contracts; diff --git a/src/modules/Elsa.Workflows.Management/Options/ManagementOptions.cs b/src/modules/Elsa.Workflows.Management/Options/ManagementOptions.cs index e6f3c02ca..86b8d332b 100644 --- a/src/modules/Elsa.Workflows.Management/Options/ManagementOptions.cs +++ b/src/modules/Elsa.Workflows.Management/Options/ManagementOptions.cs @@ -1,3 +1,4 @@ +using Elsa.Workflows.LogPersistence; using Elsa.Workflows.Management.Models; namespace Elsa.Workflows.Management.Options; diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs index 327dcc28a..df335a24d 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs @@ -109,7 +109,6 @@ public class BulkDispatchWorkflows : Activity await foreach (var item in items) { - //context.DeferTask(async () => await DispatchChildWorkflowAsync(context, item)); await DispatchChildWorkflowAsync(context, item); dispatchedInstancesCount++; } diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IActivityExecutonMapper.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IActivityExecutonMapper.cs index c20c47648..c533c2b00 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IActivityExecutonMapper.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IActivityExecutonMapper.cs @@ -10,5 +10,5 @@ public interface IActivityExecutionMapper /// /// Maps an activity execution context to an activity execution record. /// - ActivityExecutionRecord Map(ActivityExecutionContext source); + Task MapAsync(ActivityExecutionContext source); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/ILogRecordExtractor.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/ILogRecordExtractor.cs index 2a4d0564a..a5a11d350 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/ILogRecordExtractor.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/ILogRecordExtractor.cs @@ -6,5 +6,5 @@ namespace Elsa.Workflows.Runtime; public interface ILogRecordExtractor where T: ILogRecord { /// Extracts execution logs from a workflow execution context. - IEnumerable ExtractLogRecords(WorkflowExecutionContext context); + Task> ExtractLogRecordsAsync(WorkflowExecutionContext context); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Entities/ActivityExecutionRecord.cs b/src/modules/Elsa.Workflows.Runtime/Entities/ActivityExecutionRecord.cs index 6c8b1de8c..429516fa8 100644 --- a/src/modules/Elsa.Workflows.Runtime/Entities/ActivityExecutionRecord.cs +++ b/src/modules/Elsa.Workflows.Runtime/Entities/ActivityExecutionRecord.cs @@ -42,7 +42,7 @@ public class ActivityExecutionRecord : Entity, ILogRecord /// /// The state of the activity at the time this record is created or last updated. /// - public IDictionary? ActivityState { get; set; } + public IDictionary? ActivityState { get; set; } /// /// Any additional payload associated with the log record. diff --git a/src/modules/Elsa.Workflows.Runtime/Services/ActivityExecutionRecordExtractor.cs b/src/modules/Elsa.Workflows.Runtime/Services/ActivityExecutionRecordExtractor.cs index 85d1db7d5..139198419 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/ActivityExecutionRecordExtractor.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/ActivityExecutionRecordExtractor.cs @@ -6,9 +6,10 @@ namespace Elsa.Workflows.Runtime; public class ActivityExecutionRecordExtractor(IActivityExecutionMapper activityExecutionMapper) : ILogRecordExtractor { /// - public IEnumerable ExtractLogRecords(WorkflowExecutionContext context) + public async Task> ExtractLogRecordsAsync(WorkflowExecutionContext context) { var activityExecutionContexts = context.ActivityExecutionContexts; - return activityExecutionContexts.Select(activityExecutionMapper.Map).ToList(); + var tasks = activityExecutionContexts.Select(activityExecutionMapper.MapAsync).ToList(); + return await Task.WhenAll(tasks); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultActivityExecutionMapper.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultActivityExecutionMapper.cs index f6ed24440..2fe92dce7 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultActivityExecutionMapper.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultActivityExecutionMapper.cs @@ -1,5 +1,6 @@ using Elsa.Extensions; using Elsa.Workflows.Activities; +using Elsa.Workflows.LogPersistence; using Elsa.Workflows.Management.Options; using Elsa.Workflows.Models; using Elsa.Workflows.Runtime.Entities; @@ -10,14 +11,24 @@ using Microsoft.Extensions.Options; namespace Elsa.Workflows.Runtime; /// -public class DefaultActivityExecutionMapper(IOptions options) : IActivityExecutionMapper +public class DefaultActivityExecutionMapper : IActivityExecutionMapper { - private const string LogPersistenceModeKey = "logPersistenceMode"; + private readonly IOptions _options; + private readonly IDictionary _logPersistenceStrategies; + + public DefaultActivityExecutionMapper(IOptions options, ILogPersistenceStrategyService logPersistenceStrategyService) + { + _options = options; + _logPersistenceStrategies = logPersistenceStrategyService.ListStrategies().ToDictionary(x => x.GetType().GetSimpleAssemblyQualifiedName(), x => x); + } + + private const string LegacyLogPersistenceModeKey = "logPersistenceMode"; + private const string LogPersistenceStrategyKey = "logPersistenceStrategy"; /// - public ActivityExecutionRecord Map(ActivityExecutionContext source) + 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", @@ -27,42 +38,38 @@ public class DefaultActivityExecutionMapper(IOptions options) * } */ + /* The following JSON structure is expected to be found in the custom properties of the workflow and activity: + * { + * "logPersistenceStrategy": { + * "default": "null", + * "inputs": { k : v }, + * "outputs": { k: v } + * } + * } + */ + + var cancellationToken = source.WorkflowExecutionContext.CancellationToken; var workflow = (Workflow?)source.GetAncestors().FirstOrDefault(x => x.Activity is Workflow)?.Activity ?? source.WorkflowExecutionContext.Workflow; - var workflowPersistenceProperty = GetDefaultPersistenceMode(workflow.CustomProperties, () => options.Value.LogPersistenceMode); - var activityPersistenceProperties = source.Activity.CustomProperties.GetValueOrDefault>(LogPersistenceModeKey, () => new Dictionary()); - var activityPersistencePropertyDefault = GetDefaultPersistenceMode(source.Activity.CustomProperties, () => workflowPersistenceProperty); + var workflowPersistenceProperty = await GetDefaultPersistenceModeAsync(workflow.CustomProperties, () => _options.Value.LogPersistenceMode, cancellationToken); + var activityPersistencePropertyDefault = await GetDefaultPersistenceModeAsync(source.Activity.CustomProperties, () => workflowPersistenceProperty, cancellationToken); + var legacyActivityPersistenceProperties = source.Activity.CustomProperties.GetValueOrDefault>(LegacyLogPersistenceModeKey, () => new Dictionary()); + var activityPersistenceProperties = source.Activity.CustomProperties.GetValueOrDefault>(LogPersistenceStrategyKey, () => new Dictionary()); + var payload = GetPayload(source); + var outputs = GetOutputs(source); - // Get any outcomes that were added to the activity execution context. - var outcomes = source.JournalData.TryGetValue("Outcomes", out var resultValue) ? resultValue as string[] : default; - var payload = new Dictionary(); + outputs = await StorePropertyUsingPersistenceMode( + outputs, + legacyActivityPersistenceProperties!.GetValueOrDefault("outputs", () => new Dictionary())!, + activityPersistenceProperties!.GetValueOrDefault("outputs", () => new Dictionary())!, + activityPersistencePropertyDefault, + cancellationToken); - if (outcomes != null) - payload.Add("Outcomes", outcomes); - - // Get any outputs that were added to the activity execution context. - 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 != default) - return cachedValue; - - if (x.ValueGetter(activity) is Output output && source.TryGet(output.MemoryBlockReference(), out var outputValue)) - return outputValue; - - return default; - }); - - outputs = StorePropertyUsingPersistenceMode(outputs, activityPersistenceProperties!.GetValueOrDefault("outputs", () => new Dictionary())!, activityPersistencePropertyDefault); - var inputs = StorePropertyUsingPersistenceMode(source.ActivityState, activityPersistenceProperties!.GetValueOrDefault("inputs", () => new Dictionary())!, activityPersistencePropertyDefault); + var inputs = await StorePropertyUsingPersistenceMode( + source.ActivityState, + legacyActivityPersistenceProperties!.GetValueOrDefault("inputs", () => new Dictionary())!, + activityPersistenceProperties!.GetValueOrDefault("inputs", () => new Dictionary())!, + activityPersistencePropertyDefault, + cancellationToken); return new ActivityExecutionRecord { @@ -85,32 +92,63 @@ public class DefaultActivityExecutionMapper(IOptions options) }; } - private static LogPersistenceMode GetDefaultPersistenceMode(IDictionary customProperties, Func defaultFactory) + private async Task GetDefaultPersistenceModeAsync(IDictionary customProperties, Func defaultFactory, CancellationToken cancellationToken) { - var properties = customProperties.GetValueOrDefault>(LogPersistenceModeKey, () => new Dictionary()); - var persistencePropertyDefault = properties!.GetValueOrDefault("default", defaultFactory); + var legacyProperties = customProperties.GetValueOrDefault>(LegacyLogPersistenceModeKey, () => new Dictionary()); + var properties = customProperties.GetValueOrDefault>(LogPersistenceStrategyKey, () => new Dictionary()); + var defaultPersistenceStrategyTypeName = properties!.GetValueOrDefault("default"); - if (persistencePropertyDefault == LogPersistenceMode.Inherit) + if (defaultPersistenceStrategyTypeName == null) + { + var legacyPersistencePropertyDefault = legacyProperties!.GetValueOrDefault("default", defaultFactory); + + if (legacyPersistencePropertyDefault == LogPersistenceMode.Inherit) + return defaultFactory(); + return legacyPersistencePropertyDefault; + } + + var strategy = _logPersistenceStrategies.TryGetValue(defaultPersistenceStrategyTypeName, out var v) ? v : null; + + if (strategy == null) return defaultFactory(); - return persistencePropertyDefault; + + var strategyContext = new LogPersistenceStrategyContext(cancellationToken); + return await strategy.GetPersistenceModeAsync(strategyContext); } - private static Dictionary StorePropertyUsingPersistenceMode(IDictionary state, IDictionary persistenceModeConfiguration, LogPersistenceMode defaultLogPersistenceMode) + private async Task> StorePropertyUsingPersistenceMode( + IDictionary state, + IDictionary persistenceModeConfiguration, + IDictionary persistenceStrategyConfiguration, + LogPersistenceMode defaultLogPersistenceMode, + CancellationToken cancellationToken) { var result = new Dictionary(); foreach (var value in state) { - var persistence = persistenceModeConfiguration.GetValueOrDefault(value.Key.Camelize(), () => defaultLogPersistenceMode); - if (persistence.Equals(LogPersistenceMode.Include) - || (persistence.Equals(LogPersistenceMode.Inherit) && defaultLogPersistenceMode is LogPersistenceMode.Include or LogPersistenceMode.Inherit)) + var propKey = value.Key.Camelize(); + var strategy = persistenceStrategyConfiguration.GetValueOrDefault(propKey) is string strategyTypeName ? _logPersistenceStrategies.TryGetValue(strategyTypeName, out var v) ? v : null : null; + var mode = defaultLogPersistenceMode; + + if (strategy != null) + { + var strategyContext = new LogPersistenceStrategyContext(cancellationToken); + mode = await strategy.GetPersistenceModeAsync(strategyContext); + } + else + { + mode = persistenceModeConfiguration.GetValueOrDefault(propKey, () => defaultLogPersistenceMode); + } + + if (mode == LogPersistenceMode.Include || mode == LogPersistenceMode.Inherit && defaultLogPersistenceMode is LogPersistenceMode.Include or LogPersistenceMode.Inherit) result.Add(value.Key, value.Value); } return result; } - private ActivityStatus GetAggregateStatus(ActivityExecutionContext context) + private static ActivityStatus GetAggregateStatus(ActivityExecutionContext context) { // If any child activity is faulted, the aggregate status is faulted. var descendantContexts = context.GetDescendants().ToList(); @@ -120,4 +158,42 @@ public class DefaultActivityExecutionMapper(IOptions options) return context.Status; } + + private static IDictionary GetPayload(ActivityExecutionContext source) + { + var outcomes = source.JournalData.TryGetValue("Outcomes", out var resultValue) ? resultValue as string[] : default; + var payload = new Dictionary(); + + if (outcomes != null) + payload.Add("Outcomes", outcomes); + + return payload; + } + + // Get the outputs of the activity execution context. + 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 != default) + return cachedValue; + + if (x.ValueGetter(activity) is Output output && source.TryGet(output.MemoryBlockReference(), out var outputValue)) + return outputValue; + + return default; + }); + + return outputs; + } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/StoreActivityExecutionLogSink.cs b/src/modules/Elsa.Workflows.Runtime/Services/StoreActivityExecutionLogSink.cs index 9699497b0..2c280a22c 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/StoreActivityExecutionLogSink.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/StoreActivityExecutionLogSink.cs @@ -1,6 +1,7 @@ using Elsa.Mediator.Contracts; using Elsa.Workflows.Runtime.Entities; using Elsa.Workflows.Runtime.Notifications; +using Open.Linq.AsyncExtensions; namespace Elsa.Workflows.Runtime.Services; @@ -13,7 +14,7 @@ public class StoreActivityExecutionLogSink(IActivityExecutionStore activityExecu /// public async Task PersistExecutionLogsAsync(WorkflowExecutionContext context, CancellationToken cancellationToken = default) { - var records = extractor.ExtractLogRecords(context).ToList(); + var records = await extractor.ExtractLogRecordsAsync(context).ToList(); await activityExecutionStore.SaveManyAsync(records, cancellationToken); await notificationSender.SendAsync(new ActivityExecutionLogUpdated(context, records), cancellationToken); } diff --git a/src/modules/Elsa.Workflows.Runtime/Services/StoreWorkflowExecutionLogSink.cs b/src/modules/Elsa.Workflows.Runtime/Services/StoreWorkflowExecutionLogSink.cs index 353ac4c0a..a91667858 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/StoreWorkflowExecutionLogSink.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/StoreWorkflowExecutionLogSink.cs @@ -1,6 +1,7 @@ using Elsa.Mediator.Contracts; using Elsa.Workflows.Runtime.Entities; using Elsa.Workflows.Runtime.Notifications; +using Open.Linq.AsyncExtensions; namespace Elsa.Workflows.Runtime; @@ -12,7 +13,7 @@ public class StoreWorkflowExecutionLogSink(IWorkflowExecutionLogStore store, ILo /// public async Task PersistExecutionLogsAsync(WorkflowExecutionContext context, CancellationToken cancellationToken) { - var records = extractor.ExtractLogRecords(context).ToList(); + var records = await extractor.ExtractLogRecordsAsync(context).ToList(); await store.AddManyAsync(records, context.CancellationToken); await notificationSender.SendAsync(new WorkflowExecutionLogUpdated(context), context.CancellationToken); } diff --git a/src/modules/Elsa.Workflows.Runtime/Services/WorkflowExecutionLogRecordExtractor.cs b/src/modules/Elsa.Workflows.Runtime/Services/WorkflowExecutionLogRecordExtractor.cs index 4220111b5..081016122 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/WorkflowExecutionLogRecordExtractor.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/WorkflowExecutionLogRecordExtractor.cs @@ -6,9 +6,9 @@ namespace Elsa.Workflows.Runtime; public class WorkflowExecutionLogRecordExtractor(IIdentityGenerator identityGenerator) : ILogRecordExtractor { /// - public IEnumerable ExtractLogRecords(WorkflowExecutionContext context) + public Task> ExtractLogRecordsAsync(WorkflowExecutionContext context) { - return context.ExecutionLog.Select(x => new WorkflowExecutionLogRecord + var records = context.ExecutionLog.Select(x => new WorkflowExecutionLogRecord { Id = identityGenerator.GenerateId(), ActivityInstanceId = x.ActivityInstanceId, @@ -30,5 +30,7 @@ public class WorkflowExecutionLogRecordExtractor(IIdentityGenerator identityGene Timestamp = x.Timestamp, Sequence = x.Sequence }); + + return Task.FromResult(records); } } \ No newline at end of file diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/Elsa.AzureServiceBus.ComponentTests.csproj b/test/component/Elsa.AzureServiceBus.ComponentTests/Elsa.AzureServiceBus.ComponentTests.csproj index 96f81d332..89ed27dde 100644 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/Elsa.AzureServiceBus.ComponentTests.csproj +++ b/test/component/Elsa.AzureServiceBus.ComponentTests/Elsa.AzureServiceBus.ComponentTests.csproj @@ -21,6 +21,7 @@ + diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/WorkflowServer.cs b/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/WorkflowServer.cs index be10535a7..93c90530b 100644 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/WorkflowServer.cs +++ b/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/WorkflowServer.cs @@ -1,4 +1,5 @@ using System.Net.Http.Headers; +using Elsa.Agents; using Elsa.Alterations.Extensions; using Elsa.AzureServiceBus.ComponentTests.Extensions; using Elsa.EntityFrameworkCore.Extensions; @@ -75,6 +76,8 @@ public class WorkflowServer(Infrastructure infrastructure, string url) : WebAppl http.UseCache(); }); elsa.UseAzureServiceBus(); + elsa.UseAgents(); + elsa.UseAgentPersistence(feature => feature.UseEntityFrameworkCore(ef => ef.UsePostgreSql(typeof(AgentsPostgreSqlProvidersExtensions).Assembly, dbConnectionString))); }; } diff --git a/test/component/Elsa.Workflows.ComponentTests/Elsa.Workflows.ComponentTests.csproj b/test/component/Elsa.Workflows.ComponentTests/Elsa.Workflows.ComponentTests.csproj index e8dd3145c..1f285015a 100644 --- a/test/component/Elsa.Workflows.ComponentTests/Elsa.Workflows.ComponentTests.csproj +++ b/test/component/Elsa.Workflows.ComponentTests/Elsa.Workflows.ComponentTests.csproj @@ -17,6 +17,7 @@ + diff --git a/test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs b/test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs index c7d70f26c..dc868daae 100644 --- a/test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs +++ b/test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs @@ -3,6 +3,7 @@ using System.Reflection; using Elsa.Agents; using Elsa.Alterations.Extensions; using Elsa.Caching; +using Elsa.EntityFrameworkCore; using Elsa.EntityFrameworkCore.Extensions; using Elsa.EntityFrameworkCore.Modules.Alterations; using Elsa.EntityFrameworkCore.Modules.Identity; @@ -106,7 +107,7 @@ public class WorkflowServer(Infrastructure infrastructure, string url) : WebAppl http.UseCache(); }); elsa.UseAgents(); - elsa.UseAgentPersistence(feature => feature.UseEntityFrameworkCore(ef => ef.UsePostgreSql(dbConnectionString))); + elsa.UseAgentPersistence(feature => feature.UseEntityFrameworkCore(ef => ef.UsePostgreSql(typeof(AgentsPostgreSqlProvidersExtensions).Assembly, dbConnectionString))); }; }