diff --git a/PARR.API.sln b/PARR.API.sln index 9c8bcd5a..e8856eae 100644 --- a/PARR.API.sln +++ b/PARR.API.sln @@ -98,6 +98,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "PARR.Infrastructure", "PARR EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "PARR.WorkloadBuilderWorker", "PARR.WorkloadBuilderWorker\PARR.WorkloadBuilderWorker.csproj", "{835BD6CF-024E-47F5-BB02-C904AC9733D9}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "PARR.TaskReconciliationWorker", "PARR.TaskReconciliationWorker\PARR.TaskReconciliationWorker.csproj", "{DB701295-1696-4E1A-9088-D3AECF823BA6}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -274,6 +276,10 @@ Global {835BD6CF-024E-47F5-BB02-C904AC9733D9}.Debug|Any CPU.Build.0 = Debug|Any CPU {835BD6CF-024E-47F5-BB02-C904AC9733D9}.Release|Any CPU.ActiveCfg = Release|Any CPU {835BD6CF-024E-47F5-BB02-C904AC9733D9}.Release|Any CPU.Build.0 = Release|Any CPU + {DB701295-1696-4E1A-9088-D3AECF823BA6}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {DB701295-1696-4E1A-9088-D3AECF823BA6}.Debug|Any CPU.Build.0 = Debug|Any CPU + {DB701295-1696-4E1A-9088-D3AECF823BA6}.Release|Any CPU.ActiveCfg = Release|Any CPU + {DB701295-1696-4E1A-9088-D3AECF823BA6}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE diff --git a/PARR.API/Program.cs b/PARR.API/Program.cs index a4a8cca8..28a7333d 100644 --- a/PARR.API/Program.cs +++ b/PARR.API/Program.cs @@ -6,8 +6,6 @@ using PARR.API.Installers; using PARR.API.Settings; using PARR.Core; using PARR.DAL; -using PARR.Domain.Enums; -using PARR.Domain.Settings; using PARR.Infrastructure; using Serilog; using System.Reflection; diff --git a/PARR.Core/DependencyInjection.cs b/PARR.Core/DependencyInjection.cs index 80a5b702..7bbd7f2b 100644 --- a/PARR.Core/DependencyInjection.cs +++ b/PARR.Core/DependencyInjection.cs @@ -8,6 +8,7 @@ using PARR.Core.Services.Task.Handlers.Factory; using PARR.Core.Services.Task.Implementations; using PARR.Core.Services.Task.Interfaces; using PARR.Core.Services.Task.Providers; +using PARR.Core.Services.Task.ReconciliationHosted; using PARR.Core.Services.Workload.Implementations; using PARR.Core.Services.Workload.Interfaces; using PARR.Domain.Enums; @@ -108,6 +109,30 @@ namespace PARR.Core } + /// + /// Подключение сервисов для TaskReconciliation + /// + /// + /// + /// + public static IServiceCollection AddTaskReconciliation(this IServiceCollection services, IReconciliationSettings reconciliationSettings) + { + if (reconciliationSettings == null) + throw new ArgumentNullException(nameof(reconciliationSettings)); + + // Настройки + services.AddSingleton(reconciliationSettings); + + // Сервис + services.AddScoped(); + + // Воркер + services.AddHostedService(); + + return services; + } + + private static IServiceCollection AddBllMapping(this IServiceCollection services) { // AutoMapper diff --git a/PARR.Core/Services/Task/Implementations/TaskManagementService.cs b/PARR.Core/Services/Task/Implementations/TaskManagementService.cs index 636f2d52..c6b33a06 100644 --- a/PARR.Core/Services/Task/Implementations/TaskManagementService.cs +++ b/PARR.Core/Services/Task/Implementations/TaskManagementService.cs @@ -89,6 +89,10 @@ namespace PARR.Core.Services.Task.Implementations // Не пробрасываем исключение дальше — задача создана, просто не в очереди. // задача останется в БД со статусом pending, и Reconciliation Task позже ее возмет в работу logger.LogError("Не удалось опубликовать задачу {TaskId} в очередь. Задача осталась в БД со статусом Pending.", task.Id); + + // так как задачу не смогли отправить в очередь повторно, обновим ей DateModified, относительно нее считается время обработки, и задача встанет на повтор позже сама + task.DateModified = DateTimeOffset.UtcNow; + await taskRepository.CommitAsync(); } return task.Id; diff --git a/PARR.Core/Services/Task/Implementations/TaskReconciliationService.cs b/PARR.Core/Services/Task/Implementations/TaskReconciliationService.cs index 68991449..2a00d5ae 100644 --- a/PARR.Core/Services/Task/Implementations/TaskReconciliationService.cs +++ b/PARR.Core/Services/Task/Implementations/TaskReconciliationService.cs @@ -98,12 +98,15 @@ namespace PARR.Core.Services.Task.Implementations return await taskRepository.Get() .Where(t => - t.TypeCode == taskType.Code && t.DateModified < timeoutThreshold && - ( + //t.TypeCode == taskType.Code && t.DateModified < timeoutThreshold && + t.TypeCode == taskType.Code + // если это вновь созданная задача и она упала, то у нее может не быть DateModified, тогда сравним с DateCreated + && (t.DateModified ?? t.DateCreated) < timeoutThreshold + && ( // Зависла в Processing t.StatusCode == TaskItemStatusEnum.Processing //Зависли в Pending - || (t.StatusCode == TaskItemStatusEnum.Pending && t.RetryCount > 0) + || (t.StatusCode == TaskItemStatusEnum.Pending /*&& t.RetryCount > 0*/) ) ) .OrderBy(t => t.DateModified) // Сначала самые старые diff --git a/PARR.Core/Services/Task/ReconciliationHosted/ReconciliationHostedService.cs b/PARR.Core/Services/Task/ReconciliationHosted/ReconciliationHostedService.cs new file mode 100644 index 00000000..f378f1ec --- /dev/null +++ b/PARR.Core/Services/Task/ReconciliationHosted/ReconciliationHostedService.cs @@ -0,0 +1,54 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using PARR.Core.Common.Interfaces; +using PARR.Core.Services.Task.Interfaces; +using PARR.Domain.Settings; + +namespace PARR.Core.Services.Task.ReconciliationHosted +{ + /// + /// Фоновая задача для периодического запуска Reconciliation Job. + /// Обработка зависших задач. + /// + internal class ReconciliationHostedService : BackgroundService + { + // Это реализация воркера, подключается напрямую в воркер + + private readonly ILogger logger; + private readonly IReconciliationSettings reconciliationSettings; + private readonly IServiceProvider serviceProvider; + private readonly IIntervalService intervalService; + + public ReconciliationHostedService( + ILogger logger, + IReconciliationSettings reconciliationSettings, + IServiceProvider serviceProvider, + IIntervalService intervalService + ) + { + this.logger = logger; + this.reconciliationSettings = reconciliationSettings; + this.serviceProvider = serviceProvider; + this.intervalService = intervalService; + } + + protected override async System.Threading.Tasks.Task ExecuteAsync(CancellationToken stoppingToken) + { + await intervalService.IntervalInitAsync(async () => + { + await using (var scope = serviceProvider.CreateAsyncScope()) + { + var reconciliationService = scope.ServiceProvider.GetRequiredService(); + + // Запускаем проверку + var report = await reconciliationService.RunAsync(); + + if (report.HasChanged) + logger.LogInformation("Reconciliation Job выполнен. Проверено: {Checked}, Исправлено: {Fixed}", report.CheckedCount, report.FixedCount); + } + + }, reconciliationSettings.CheckInterval); + } + } +} diff --git a/PARR.Core/Services/Workload/Implementations/WorkloadCacheBuilderService.cs b/PARR.Core/Services/Workload/Implementations/WorkloadCacheBuilderService.cs index 57715328..14c25621 100644 --- a/PARR.Core/Services/Workload/Implementations/WorkloadCacheBuilderService.cs +++ b/PARR.Core/Services/Workload/Implementations/WorkloadCacheBuilderService.cs @@ -2,7 +2,6 @@ using Microsoft.Extensions.Logging; using PARR.Core.Common.Interfaces; using PARR.Core.Common.Interfaces.RabbitServices; -using PARR.Core.Services.Task.Handlers; using PARR.Core.Services.Task.Handlers.Factory; using PARR.Core.Services.Workload.Interfaces; using PARR.Domain.Common.Rabbit.Messages; diff --git a/PARR.TaskReconciliationWorker/PARR.TaskReconciliationWorker.csproj b/PARR.TaskReconciliationWorker/PARR.TaskReconciliationWorker.csproj new file mode 100644 index 00000000..b4d2b4c6 --- /dev/null +++ b/PARR.TaskReconciliationWorker/PARR.TaskReconciliationWorker.csproj @@ -0,0 +1,25 @@ + + + + net7.0 + enable + enable + dotnet-PARR.TaskReconciliationWorker-90386488-d496-4f4d-8dcd-7dcc50882876 + + + + + + + + + + + + + + + + + + diff --git a/PARR.TaskReconciliationWorker/Program.cs b/PARR.TaskReconciliationWorker/Program.cs new file mode 100644 index 00000000..734837da --- /dev/null +++ b/PARR.TaskReconciliationWorker/Program.cs @@ -0,0 +1,49 @@ +using Elastic.CommonSchema.Serilog; +using PARR.Core; +using PARR.DAL; +using PARR.Infrastructure; +using PARR.TaskReconciliationWorker.Settings; +using Serilog; + +var builder = Host.CreateApplicationBuilder(); + +builder.Services.AddLogging(config => +{ + config.ClearProviders(); + + var logger = new LoggerConfiguration(); + + if (builder.Environment.IsProduction()) + logger.WriteTo.Console(new EcsTextFormatter()); + else + logger.WriteTo.Console(); + + logger.ReadFrom.Configuration(builder.Configuration); + + config.AddSerilog(logger.CreateLogger()); +}); + +// --- Настройки --- +var workerSettings = new WorkerSettings(); +builder.Configuration.GetSection(nameof(WorkerSettings)).Bind(workerSettings); + +builder.Services.InstallDalServices(builder.Configuration); +builder.Services.AddCoreServices(builder.Configuration); +builder.Services.AddInfrastructureServices(builder.Configuration); +builder.Configuration.AddDalConfigurations(builder.Services); +builder.Services.AddDallSettings(builder.Configuration); + +builder.Services.AddTaskManagement((provider, taskType) => +{ + // настройки очередей + //var workerSettings = provider.GetRequiredService(); + if (workerSettings.MqSettings.Tasks.TryGetValue(taskType, out var settings)) + return settings; + + throw new ArgumentException($"Настройки для типа задания {taskType} не найдены в конфигурации."); +}); + +builder.Services.AddTaskReconciliation(workerSettings); + +var host = builder.Build(); +host.Run(); \ No newline at end of file diff --git a/PARR.TaskReconciliationWorker/Properties/launchSettings.json b/PARR.TaskReconciliationWorker/Properties/launchSettings.json new file mode 100644 index 00000000..db49dcf0 --- /dev/null +++ b/PARR.TaskReconciliationWorker/Properties/launchSettings.json @@ -0,0 +1,11 @@ +{ + "profiles": { + "PARR.TaskReconciliationWorker": { + "commandName": "Project", + "dotnetRunMessages": true, + "environmentVariables": { + "DOTNET_ENVIRONMENT": "Development" + } + } + } +} diff --git a/PARR.TaskReconciliationWorker/Settings/WorkerSettings.cs b/PARR.TaskReconciliationWorker/Settings/WorkerSettings.cs new file mode 100644 index 00000000..16976a84 --- /dev/null +++ b/PARR.TaskReconciliationWorker/Settings/WorkerSettings.cs @@ -0,0 +1,18 @@ +using PARR.Domain.Enums; +using PARR.Domain.Settings; + +namespace PARR.TaskReconciliationWorker.Settings +{ + internal class WorkerSettings : IReconciliationSettings + { + public TimeSpan CheckInterval { get; set; } = TimeSpan.FromMinutes(10); + public int MaxTaskPerRun { get; set; } = 100; + + public MqSettings MqSettings { get; set; } = new(); + } + + internal class MqSettings + { + public Dictionary Tasks { get; set; } = new(); + } +} diff --git a/PARR.TaskReconciliationWorker/appsettings.Development.json b/PARR.TaskReconciliationWorker/appsettings.Development.json new file mode 100644 index 00000000..7e71cd41 --- /dev/null +++ b/PARR.TaskReconciliationWorker/appsettings.Development.json @@ -0,0 +1,38 @@ +{ + "ConnectionStrings": { + "RedisConnection": "10.99.253.216:6379,password=ParrP@ssPtk202MMdevDvs" + }, + "Logging": { + "LogLevel": { + "Default": "Debug", + "Microsoft.Hosting.Lifetime": "Information" + } + }, + "Serilog": { + "MinimumLevel": { + "Default": "Debug", + "Override": { + "Microsoft": "Warning", + "Microsoft.Hosting.Lifetime": "Information" + } + }, + "WriteTo": [ + { + "Name": "File", + "Args": { + "path": "log/log-.txt", + "rollingInterval": "Day" + } + } + ] + }, + "WorkerSettings": { + "MqSettings": { + "Tasks": { + "Workload": { + "HostName": "10.99.253.216" + } + } + } + } +} diff --git a/PARR.TaskReconciliationWorker/appsettings.json b/PARR.TaskReconciliationWorker/appsettings.json new file mode 100644 index 00000000..05d4b3fe --- /dev/null +++ b/PARR.TaskReconciliationWorker/appsettings.json @@ -0,0 +1,39 @@ +{ + "ConnectionStrings": { + "DefaultConnection": "Server=10.99.253.184;Database=parr;User Id=app_parr; Password=PosdfkhT&)%sdfligL&%5546;", + "RedisConnection": "parr-redis:6379,password=ParrP@ssPtk202MMdevDvs" + }, + "Logging": { + "LogLevel": { + "Default": "Information", + "Microsoft.EntityFrameworkCore": "Error", + "Microsoft.EntityFrameworkCore.Database.Command": "Warning", + "Microsoft.AspNetCore": "Warning" + } + }, + "Serilog": { + "MinimumLevel": { + "Default": "Information", + "Override": { + "Microsoft": "Warning", + "Microsoft.EntityFrameworkCore": "Error", + "Microsoft.EntityFrameworkCore.Database.Command": "Warning", + "Microsoft.Hosting.Lifetime": "Information" + } + } + }, + "WorkerSettings": { + "CheckInterval": "00:10:00", + "MaxTaskPerRun": 100, + "MqSettings": { + "Tasks": { + "Workload": { + "HostName": "parr-rabbitmq", + "QueueName": "parr-task-workload", + "User": "task_workload_writer", + "Password": "UFtsduifytIUR$^85382Wt1few" + } + } + } + } +}