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

Optional rollover strategy

Browse source
This commit is contained in:
Gürkan Güran 2023-01-10 14:29:27 +01:00
parent ce2434f9fd
commit 02ff020b6e
20 changed files with 233 additions and 158 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
src/modules/Elsa.Elasticsearch/Common/PersistanceFeatureBase.cs Normal file
View file

@ -0,0 +1,27 @@
using Elasticsearch.Net;
using Elsa.Elasticsearch.Extensions;
using Elsa.Elasticsearch.HostedServices;
using Elsa.Elasticsearch.Models;
using Elsa.Elasticsearch.Options;
using Elsa.Elasticsearch.Scheduling;
using Elsa.Elasticsearch.Services;
using Elsa.Features.Abstractions;
using Elsa.Features.Services;
using Microsoft.Extensions.DependencyInjection;
using Nest;
namespace Elsa.Elasticsearch.Common;
public abstract class ElasticPersistanceFeatureBase : FeatureBase
{
public ElasticPersistanceFeatureBase(IModule module) : base(module)
{
}
protected void AddStore<TModel, TStore>() where TModel : class where TStore : class
{
Services
.AddSingleton<ElasticStore<TModel>>()
.AddSingleton<TStore>();
}
}

7
src/modules/Elsa.Elasticsearch/Extensions/ElasticExtensions.cs
View file

@ -1,6 +1,5 @@
using Elasticsearch.Net;
using Elsa.Elasticsearch.Common;
using Elsa.Elasticsearch.Implementations.RolloverStrategies;
using Elsa.Elasticsearch.Models;
using Elsa.Elasticsearch.Options;
using Elsa.Elasticsearch.Services;
@ -36,9 +35,9 @@ public static class ElasticExtensions
return settings;
}
public static void ApplyRolloverStrategy(this ElasticClient client, IDictionary<Type,string> aliasConfig, IndexRolloverStrategy strategy)
public static void ConfigureAliasNaming(this ElasticClient client, IDictionary<Type,string> aliasConfig, IndexRolloverStrategy strategy)
{
var strategyInstance = (IRolloverStrategy)Activator.CreateInstance(strategy.Value, args: client)!;
strategyInstance.Apply(Utils.GetElasticDocumentTypes(), aliasConfig);
var namingStrategy = (IIndexNamingStrategy)Activator.CreateInstance(strategy.IndexNamingStrategy, args: client)!;
namingStrategy.Apply(Utils.GetElasticDocumentTypes(), aliasConfig);
}
}

31
src/modules/Elsa.Elasticsearch/Extensions/ModuleExtensions.cs Normal file
View file

@ -0,0 +1,31 @@
using Elsa.Elasticsearch.Common;
using Elsa.Elasticsearch.Features;
using Elsa.Elasticsearch.Models;
using Elsa.Elasticsearch.Options;
using Elsa.Features.Services;
namespace Elsa.Elasticsearch.Extensions;
public static class ModuleExtensions
{
/// <summary>
/// Enables the <see cref="ElasticsearchFeature"/> feature.
/// </summary>
public static IModule UseElasticsearch(
this IModule module,
ElasticsearchOptions options,
IndexRolloverStrategy? rolloverStrategy = default,
IDictionary<string,string>? indexConfig = default,
Action<ElasticsearchFeature>? configure = default)
{
configure += f =>
{
f.Options = options;
f.IndexRolloverStrategy = rolloverStrategy;
f.IndexConfig = Utils.ResolveAliasConfig(f.IndexConfig, options.IndexConfig, indexConfig);
};
module.Configure(configure);
return module;
}
}

29
src/modules/Elsa.Elasticsearch/Common/ElasticFeatureBase.cs → src/modules/Elsa.Elasticsearch/Features/ElasticsearchFeature.cs
View file

@ -1,20 +1,18 @@
using Elasticsearch.Net;
using Elsa.Elasticsearch.Extensions;
using Elsa.Elasticsearch.HostedServices;
using Elsa.Elasticsearch.Models;
using Elsa.Elasticsearch.Options;
using Elsa.Elasticsearch.Scheduling;
using Elsa.Elasticsearch.Services;
using Elsa.Features.Abstractions;
using Elsa.Features.Services;
using Microsoft.Extensions.DependencyInjection;
using Nest;
namespace Elsa.Elasticsearch.Common;
namespace Elsa.Elasticsearch.Features;
public abstract class ElasticFeatureBase : FeatureBase
public class ElasticsearchFeature : FeatureBase
{
public ElasticFeatureBase(IModule module) : base(module)
public ElasticsearchFeature(IModule module) : base(module)
{
}
@ -24,18 +22,24 @@ public abstract class ElasticFeatureBase : FeatureBase
public override void ConfigureHostedServices()
{
Module.ConfigureHostedService<ConfigureElasticsearchHostedService>(-1);
Module.ConfigureHostedService<ConfigureMappingHostedService>(-1);
if (IndexRolloverStrategy != null)
{
Module.ConfigureHostedService<ConfigureIndexRolloverHostedService>(-1);
}
}
public override void Apply()
{
if (Services.Any(x => x.ServiceType == typeof(ElasticClient))) return;
var elasticClient = new ElasticClient(GetSettings());
if (IndexRolloverStrategy != null)
{
elasticClient.ApplyRolloverStrategy(IndexConfig, IndexRolloverStrategy!);
elasticClient.ConfigureAliasNaming(IndexConfig, IndexRolloverStrategy);
var typeInstance = (IIndexRolloverStrategy) Activator.CreateInstance(IndexRolloverStrategy.Value, args: elasticClient)!;
Services.AddSingleton<IIndexRolloverStrategy>(_ => typeInstance);
}
Services.AddSingleton(elasticClient);
@ -47,11 +51,4 @@ public abstract class ElasticFeatureBase : FeatureBase
.ConfigureAuthentication(Options)
.ConfigureMapping(IndexConfig);
}
protected void AddStore<TModel, TStore>() where TModel : class where TStore : class
{
Services
.AddSingleton<ElasticStore<TModel>>()
.AddSingleton<TStore>();
}
}

51
src/modules/Elsa.Elasticsearch/HostedServices/ConfigureElasticsearchHostedService.cs
View file

@ -1,51 +0,0 @@
using Elsa.Elasticsearch.Scheduling;
using Elsa.Jobs.Schedules;
using Elsa.Jobs.Services;
using Elsa.Workflows.Management.Entities;
using Microsoft.Extensions.Hosting;
using Nest;
namespace Elsa.Elasticsearch.HostedServices;
public class ConfigureElasticsearchHostedService : IHostedService
{
private readonly ElasticClient _elasticClient;
private readonly IJobScheduler _jobScheduler;
public ConfigureElasticsearchHostedService(ElasticClient elasticClient, IJobScheduler jobScheduler)
{
_elasticClient = elasticClient;
_jobScheduler = jobScheduler;
}
public async Task StartAsync(CancellationToken cancellationToken)
{
await FlattenProperties(cancellationToken);
await ScheduleIndexAndAliasConfiguration(cancellationToken);
}
private async Task ScheduleIndexAndAliasConfiguration(CancellationToken cancellationToken)
{
var job = new ConfigureElasticIndicesJob();
var schedule = new CronSchedule
{
// At the start of every month
CronExpression = "*/5 * * * *"
};
await _jobScheduler.ScheduleAsync(job, GetType().Name, schedule, cancellationToken: cancellationToken);
}
private async Task FlattenProperties(CancellationToken cancellationToken)
{
await _elasticClient.Indices.PutMappingAsync<WorkflowInstance>(
descriptor => descriptor
.Properties(p => p
.Flattened(d => d
.Name(p => p.WorkflowState.Properties))),
cancellationToken);
}
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
}

31
src/modules/Elsa.Elasticsearch/HostedServices/ConfigureIndexRolloverHostedService.cs Normal file
View file

@ -0,0 +1,31 @@
using Elsa.Elasticsearch.Scheduling;
using Elsa.Jobs.Schedules;
using Elsa.Jobs.Services;
using Microsoft.Extensions.Hosting;
namespace Elsa.Elasticsearch.HostedServices;
public class ConfigureIndexRolloverHostedService : IHostedService
{
private readonly IJobScheduler _jobScheduler;
public ConfigureIndexRolloverHostedService(IJobScheduler jobScheduler)
{
_jobScheduler = jobScheduler;
}
public async Task StartAsync(CancellationToken cancellationToken)
{
var job = new ConfigureIndexRolloverJob();
var schedule = new CronSchedule
{
// At the beginning of every month
//CronExpression = "0 0 1 * *"
CronExpression = "*/5 * * * *"
};
await _jobScheduler.ScheduleAsync(job, GetType().Name, schedule, cancellationToken: cancellationToken);
}
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
}

27
src/modules/Elsa.Elasticsearch/HostedServices/ConfigureMappingHostedService.cs Normal file
View file

@ -0,0 +1,27 @@
using Elsa.Workflows.Management.Entities;
using Microsoft.Extensions.Hosting;
using Nest;
namespace Elsa.Elasticsearch.HostedServices;
public class ConfigureMappingHostedService : IHostedService
{
private readonly ElasticClient _elasticClient;
public ConfigureMappingHostedService(ElasticClient elasticClient)
{
_elasticClient = elasticClient;
}
public async Task StartAsync(CancellationToken cancellationToken)
{
await _elasticClient.Indices.PutMappingAsync<WorkflowInstance>(
descriptor => descriptor
.Properties(p => p
.Flattened(d => d
.Name(p => p.WorkflowState.Properties))),
cancellationToken);
}
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
}

35
src/modules/Elsa.Elasticsearch/Implementations/IndexNamingStrategies/NamingWithYearAndMonth.cs Normal file
View file

@ -0,0 +1,35 @@
using Elsa.Elasticsearch.Common;
using Elsa.Elasticsearch.Services;
using Nest;
namespace Elsa.Elasticsearch.Implementations.IndexNamingStrategies;
public class NamingWithYearAndMonth : IIndexNamingStrategy
{
private readonly ElasticClient _client;
public NamingWithYearAndMonth(ElasticClient client)
{
_client = client;
}
public void Apply(IEnumerable<Type> typesToConfigure, IDictionary<Type, string> aliasConfig)
{
foreach (var type in typesToConfigure)
{
var aliasName = aliasConfig[type];
var indexName = Utils.GenerateIndexName(aliasName);
var indexExists = _client.Indices.Exists(indexName).Exists;
if (indexExists) continue;
var response = _client.Indices.Create(indexName, s => s
.Aliases(a => a.Alias(aliasName))
.Map(m => m.AutoMap(type)));
if (response.IsValid) continue;
throw response.OriginalException;
}
}
}

17
src/modules/Elsa.Elasticsearch/Models/IndexNamingStrategy.cs Normal file
View file

@ -0,0 +1,17 @@
using Elsa.Elasticsearch.Implementations.IndexNamingStrategies;
namespace Elsa.Elasticsearch.Models;
public class IndexNamingStrategy
{
private IndexNamingStrategy(Type value) { Value = value; }
public Type Value { get; private set; }
public static IndexNamingStrategy NamingWithYearAndMonth => new (typeof(NamingWithYearAndMonth));
public override string ToString()
{
return Value.Name;
}
}

17
src/modules/Elsa.Elasticsearch/Models/IndexRolloverStrategy.cs
View file

@ -1,17 +1,20 @@
using System.Reflection.Emit;
using Elsa.Elasticsearch.Implementations.IndexNamingStrategies;
using Elsa.Elasticsearch.Implementations.RolloverStrategies;
using Elsa.Elasticsearch.Services;
namespace Elsa.Elasticsearch.Models;
public class IndexRolloverStrategy
{
private IndexRolloverStrategy(Type value) { Value = value; }
private IndexRolloverStrategy(Type value, Type indexNamingStrategy)
{
Value = value;
IndexNamingStrategy = indexNamingStrategy;
}
public Type Value { get; private set; }
public Type IndexNamingStrategy { get; private set; }
public static IndexRolloverStrategy RolloverOnMonthlyBasis => new (typeof(RolloverOnMonthlyBasis));
public override string ToString()
{
return Value.Name;
}
public static IndexRolloverStrategy RolloverOnMonthlyBasis => new (typeof(RolloverOnMonthlyBasis),typeof(NamingWithYearAndMonth));
}

2
src/modules/Elsa.Elasticsearch/Modules/Management/ElasticWorkflowInstanceFeature.cs
View file

@ -8,7 +8,7 @@ using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Elasticsearch.Modules.Management;
[DependsOn(typeof(WorkflowManagementFeature))]
public class ElasticWorkflowInstanceFeature : ElasticFeatureBase
public class ElasticWorkflowInstanceFeature : ElasticPersistanceFeatureBase
{
public ElasticWorkflowInstanceFeature(IModule module) : base(module)
{

17
src/modules/Elsa.Elasticsearch/Modules/Management/Extensions.cs
View file

@ -1,6 +1,3 @@
using Elsa.Elasticsearch.Common;
using Elsa.Elasticsearch.Models;
using Elsa.Elasticsearch.Options;
using Elsa.Workflows.Management.Features;
namespace Elsa.Elasticsearch.Modules.Management;
@ -10,20 +7,8 @@ public static class Extensions
/// <summary>
/// Configures the <see cref="WorkflowInstanceFeature"/> to use the <see cref="ElasticWorkflowInstanceFeature"/>.
/// </summary>
public static WorkflowInstanceFeature UseElasticsearch(
this WorkflowInstanceFeature feature,
ElasticsearchOptions options,
IndexRolloverStrategy? rolloverStrategy = default,
IDictionary<string,string>? indexConfig = default,
Action<ElasticWorkflowInstanceFeature>? configure = default)
public static WorkflowInstanceFeature UseElasticsearch(this WorkflowInstanceFeature feature, Action<ElasticWorkflowInstanceFeature>? configure = default)
{
configure += f =>
{
f.Options = options;
f.IndexRolloverStrategy = rolloverStrategy;
f.IndexConfig = Utils.ResolveAliasConfig(f.IndexConfig, options.IndexConfig, indexConfig);
};
feature.Module.Configure(configure);
return feature;
}

2
src/modules/Elsa.Elasticsearch/Modules/Runtime/ElasticExecutionLogRecordFeature.cs
View file

@ -8,7 +8,7 @@ using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Elasticsearch.Modules.Runtime;
[DependsOn(typeof(WorkflowRuntimeFeature))]
public class ElasticExecutionLogRecordFeature : ElasticFeatureBase
public class ElasticExecutionLogRecordFeature : ElasticPersistanceFeatureBase
{
public ElasticExecutionLogRecordFeature(IModule module) : base(module)
{

17
src/modules/Elsa.Elasticsearch/Modules/Runtime/Extensions.cs
View file

@ -1,6 +1,3 @@
using Elsa.Elasticsearch.Common;
using Elsa.Elasticsearch.Models;
using Elsa.Elasticsearch.Options;
using Elsa.Workflows.Runtime.Features;
namespace Elsa.Elasticsearch.Modules.Runtime;
@ -10,20 +7,8 @@ public static class Extensions
/// <summary>
/// Configures the <see cref="ExecutionLogRecordFeature"/> to use the <see cref="ElasticExecutionLogRecordFeature"/>.
/// </summary>
public static ExecutionLogRecordFeature UseElasticsearch(
this ExecutionLogRecordFeature feature,
ElasticsearchOptions options,
IndexRolloverStrategy? rolloverStrategy = default,
IDictionary<string,string>? indexConfig = default,
Action<ElasticExecutionLogRecordFeature>? configure = default)
public static ExecutionLogRecordFeature UseElasticsearch(this ExecutionLogRecordFeature feature, Action<ElasticExecutionLogRecordFeature>? configure = default)
{
configure += f =>
{
f.Options = options;
f.IndexRolloverStrategy = rolloverStrategy;
f.IndexConfig = Utils.ResolveAliasConfig(f.IndexConfig, options.IndexConfig, indexConfig);
};
feature.Module.Configure(configure);
return feature;
}

39
src/modules/Elsa.Elasticsearch/Scheduling/ConfigureElasticIndicesJob.cs
View file

@ -1,39 +0,0 @@
using System.Text.Json.Serialization;
using Elsa.Elasticsearch.Common;
using Elsa.Jobs.Models;
using Nest;
using Job = Elsa.Jobs.Abstractions.Job;
namespace Elsa.Elasticsearch.Scheduling;
public class ConfigureElasticIndicesJob : Job
{
[JsonConstructor]
public ConfigureElasticIndicesJob()
{
}
protected override async ValueTask ExecuteAsync(JobExecutionContext context)
{
var client = context.GetRequiredService<ElasticClient>();
var indexAliasGroups = await client.Indices.GetAliasAsync();
foreach (var group in indexAliasGroups.Indices)
{
// Only 1 alias exists per index in Elsa Elasticsearch configuration
var aliasPointingCurrentIndex = group.Value.Aliases.Keys.Single();
var currentIndexName = group.Key.Name;
var newIndexName = Utils.GenerateIndexName(aliasPointingCurrentIndex);
var indexExists = (await client.Indices.ExistsAsync(newIndexName)).Exists;
if (indexExists) continue;
// Point the alias to the new writable index
await client.Indices.BulkAliasAsync(aliases => aliases
.Remove(a => a.Alias(aliasPointingCurrentIndex).Index(currentIndexName))
.Add(a => a.Alias(aliasPointingCurrentIndex).Index(currentIndexName).IsWriteIndex(false))
.Add(a => a.Alias(aliasPointingCurrentIndex).Index(newIndexName).IsWriteIndex()));
}
}
}

22
src/modules/Elsa.Elasticsearch/Scheduling/ConfigureIndexRolloverJob.cs Normal file
View file

@ -0,0 +1,22 @@
using System.Text.Json.Serialization;
using Elsa.Elasticsearch.Common;
using Elsa.Elasticsearch.Services;
using Elsa.Jobs.Models;
using Microsoft.Extensions.DependencyInjection;
using Job = Elsa.Jobs.Abstractions.Job;
namespace Elsa.Elasticsearch.Scheduling;
public class ConfigureIndexRolloverJob : Job
{
[JsonConstructor]
public ConfigureIndexRolloverJob()
{
}
protected override async ValueTask ExecuteAsync(JobExecutionContext context)
{
var rolloverStrategy = context.ServiceProvider.GetRequiredService<IIndexRolloverStrategy>();
await rolloverStrategy.ApplyAsync(Utils.GetElasticDocumentTypes(), context.CancellationToken);
}
}

6
src/modules/Elsa.Elasticsearch/Services/IIndexNamingStrategy.cs Normal file
View file

@ -0,0 +1,6 @@
namespace Elsa.Elasticsearch.Services;
public interface IIndexNamingStrategy
{
void Apply(IEnumerable<Type> typesToConfigure, IDictionary<Type, string> aliasConfig);
}

6
src/modules/Elsa.Elasticsearch/Services/IIndexRolloverStrategy.cs Normal file
View file

@ -0,0 +1,6 @@
namespace Elsa.Elasticsearch.Services;
public interface IIndexRolloverStrategy
{
Task ApplyAsync(IEnumerable<Type> types, CancellationToken cancellationToken = default);
}

6
src/modules/Elsa.Elasticsearch/Services/IRolloverStrategy.cs
View file

@ -1,6 +0,0 @@
namespace Elsa.Elasticsearch.Services;
public interface IRolloverStrategy
{
void Apply(IEnumerable<Type> types, IDictionary<Type,string> aliasConfig);
}

2
src/modules/Elsa.Quartz/Extensions/DependencyInjectionExtensions.cs
View file

@ -13,7 +13,7 @@ public static class DependencyInjectionExtensions
{
quartz.AddJob<RunWorkflowJob>();
quartz.AddJob<ResumeWorkflowJob>();
quartz.AddJob<ConfigureElasticIndicesJob>();
quartz.AddJob<ConfigureIndexRolloverJob>();
return quartz;
}