From fc8d8d92b284d55035cdca1952a8ab5c92a64390 Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Fri, 24 Jan 2025 15:50:28 -0600 Subject: [PATCH] Support distributed crontab workers. --- .../BotSharp.Abstraction.csproj | 2 +- .../Rules}/IRuleTrigger.cs | 2 +- .../Abstraction/ICrontabHook.cs | 3 + .../BotSharp.Core.Crontab/CrontabPlugin.cs | 1 + .../Services/CrontabEventSubscription.cs | 59 +++++++++++++++++++ .../Services/CrontabService.cs | 11 ++-- .../Services/CrontabWatcher.cs | 20 +++++-- .../Engines/IRuleEngine.cs | 2 - .../BotSharp.Core.Rules/Engines/RuleEngine.cs | 3 - .../BotSharp.Core.Rules/Using.cs | 4 +- .../Infrastructures/Events/RedisPublisher.cs | 6 +- .../Controllers/RulesController.cs | 1 + .../Functions/HandleHttpRequestFn.cs | 5 ++ 13 files changed, 99 insertions(+), 20 deletions(-) rename src/Infrastructure/{BotSharp.Core.Rules/Triggers => BotSharp.Abstraction/Rules}/IRuleTrigger.cs (85%) create mode 100644 src/Infrastructure/BotSharp.Core.Crontab/Services/CrontabEventSubscription.cs diff --git a/src/Infrastructure/BotSharp.Abstraction/BotSharp.Abstraction.csproj b/src/Infrastructure/BotSharp.Abstraction/BotSharp.Abstraction.csproj index 0e69d7b6..900bc4a7 100644 --- a/src/Infrastructure/BotSharp.Abstraction/BotSharp.Abstraction.csproj +++ b/src/Infrastructure/BotSharp.Abstraction/BotSharp.Abstraction.csproj @@ -1,4 +1,4 @@ - + $(TargetFramework) diff --git a/src/Infrastructure/BotSharp.Core.Rules/Triggers/IRuleTrigger.cs b/src/Infrastructure/BotSharp.Abstraction/Rules/IRuleTrigger.cs similarity index 85% rename from src/Infrastructure/BotSharp.Core.Rules/Triggers/IRuleTrigger.cs rename to src/Infrastructure/BotSharp.Abstraction/Rules/IRuleTrigger.cs index dbf88331..f5593735 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/Triggers/IRuleTrigger.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Rules/IRuleTrigger.cs @@ -1,4 +1,4 @@ -namespace BotSharp.Core.Rules.Triggers; +namespace BotSharp.Abstraction.Rules; public interface IRuleTrigger { diff --git a/src/Infrastructure/BotSharp.Core.Crontab/Abstraction/ICrontabHook.cs b/src/Infrastructure/BotSharp.Core.Crontab/Abstraction/ICrontabHook.cs index bf4e5868..45f997e0 100644 --- a/src/Infrastructure/BotSharp.Core.Crontab/Abstraction/ICrontabHook.cs +++ b/src/Infrastructure/BotSharp.Core.Crontab/Abstraction/ICrontabHook.cs @@ -2,6 +2,9 @@ namespace BotSharp.Core.Crontab.Abstraction; public interface ICrontabHook { + string[]? Triggers + => null; + Task OnCronTriggered(CrontabItem item) => Task.CompletedTask; diff --git a/src/Infrastructure/BotSharp.Core.Crontab/CrontabPlugin.cs b/src/Infrastructure/BotSharp.Core.Crontab/CrontabPlugin.cs index 26f5cfad..9e220354 100644 --- a/src/Infrastructure/BotSharp.Core.Crontab/CrontabPlugin.cs +++ b/src/Infrastructure/BotSharp.Core.Crontab/CrontabPlugin.cs @@ -34,5 +34,6 @@ public class CrontabPlugin : IBotSharpPlugin services.AddScoped(); services.AddScoped(); services.AddHostedService(); + services.AddHostedService(); } } diff --git a/src/Infrastructure/BotSharp.Core.Crontab/Services/CrontabEventSubscription.cs b/src/Infrastructure/BotSharp.Core.Crontab/Services/CrontabEventSubscription.cs new file mode 100644 index 00000000..fa3b372d --- /dev/null +++ b/src/Infrastructure/BotSharp.Core.Crontab/Services/CrontabEventSubscription.cs @@ -0,0 +1,59 @@ +using BotSharp.Abstraction.Infrastructures.Events; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using System.Runtime.InteropServices; + +namespace BotSharp.Core.Crontab.Services; + +public class CrontabEventSubscription : BackgroundService +{ + private readonly ILogger _logger; + private readonly IServiceProvider _services; + + public CrontabEventSubscription(IServiceProvider services, ILogger logger) + { + _logger = logger; + _services = services; + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + _logger.LogInformation("Crontab event subscription background service is running."); + + using (var scope = _services.CreateScope()) + { + var subscriber = scope.ServiceProvider.GetRequiredService(); + var cron = scope.ServiceProvider.GetRequiredService(); + var crons = await cron.GetCrontable(); + foreach (var item in crons) + { + _ = Task.Run(async () => + { + await subscriber.SubscribeAsync($"Crontab:{item.Title}", + "Crontab", + port: 0, + priorityEnabled: false, async (sender, args) => + { + var scope = _services.CreateScope(); + cron = scope.ServiceProvider.GetRequiredService(); + await cron.ScheduledTimeArrived(item); + }, stoppingToken: stoppingToken); + }); + } + } + + + /*using (var scope = _services.CreateScope()) + { + var cron = scope.ServiceProvider.GetRequiredService(); + var crons = await cron.GetCrontable(); + + + + while (!stoppingToken.IsCancellationRequested) + { + await Task.Delay(1000, stoppingToken); + } + }*/ + } +} diff --git a/src/Infrastructure/BotSharp.Core.Crontab/Services/CrontabService.cs b/src/Infrastructure/BotSharp.Core.Crontab/Services/CrontabService.cs index 70d16a0b..e2ddc4bf 100644 --- a/src/Infrastructure/BotSharp.Core.Crontab/Services/CrontabService.cs +++ b/src/Infrastructure/BotSharp.Core.Crontab/Services/CrontabService.cs @@ -55,12 +55,15 @@ public class CrontabService : ICrontabService public async Task ScheduledTimeArrived(CrontabItem item) { _logger.LogDebug($"ScheduledTimeArrived {item}"); - + await HookEmitter.Emit(_services, async hook => { - await hook.OnTaskExecuting(item); - await hook.OnCronTriggered(item); - await hook.OnTaskExecuted(item); + if (hook.Triggers == null || hook.Triggers.Contains(item.Title)) + { + await hook.OnTaskExecuting(item); + await hook.OnCronTriggered(item); + await hook.OnTaskExecuted(item); + } }); } } diff --git a/src/Infrastructure/BotSharp.Core.Crontab/Services/CrontabWatcher.cs b/src/Infrastructure/BotSharp.Core.Crontab/Services/CrontabWatcher.cs index 7f47c3f0..f936c172 100644 --- a/src/Infrastructure/BotSharp.Core.Crontab/Services/CrontabWatcher.cs +++ b/src/Infrastructure/BotSharp.Core.Crontab/Services/CrontabWatcher.cs @@ -1,4 +1,5 @@ using BotSharp.Abstraction.Infrastructures; +using BotSharp.Abstraction.Infrastructures.Events; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using NCrontab; @@ -26,9 +27,9 @@ public class CrontabWatcher : BackgroundService while (!stoppingToken.IsCancellationRequested) { - var delay = Task.Delay(1000 * 10, stoppingToken); + var delay = Task.Delay(1000, stoppingToken); - await locker.LockAsync("CrontabWatcher", async () => + await locker.LockAsync("CrontabWatcher:locker", async () => { await RunCronChecker(scope.ServiceProvider); }); @@ -44,8 +45,13 @@ public class CrontabWatcher : BackgroundService { var cron = services.GetRequiredService(); var crons = await cron.GetCrontable(); + + var publisher = services.GetRequiredService(); + foreach (var item in crons) { + _logger.LogDebug($"[{DateTime.UtcNow}] Cron task ({item.Title}, {item.Cron}), Last Execution Time: {item.LastExecutionTime}"); + try { // strip seconds from cron expression @@ -80,8 +86,10 @@ public class CrontabWatcher : BackgroundService if (matches) { - _logger.LogDebug($"The current time matches the cron expression {item}"); - cron.ScheduledTimeArrived(item); + _logger.LogInformation($"The current time matches the cron expression {item}"); + + await publisher.PublishAsync($"Crontab:{item.Title}", item.Cron); + // cron.ScheduledTimeArrived(item); } } catch (Exception ex) @@ -97,9 +105,9 @@ public class CrontabWatcher : BackgroundService var nextOccurrence = schedule.GetNextOccurrence(DateTime.UtcNow); var afterNextOccurrence = schedule.GetNextOccurrence(nextOccurrence); var interval = afterNextOccurrence - nextOccurrence; - if (interval.TotalMinutes < 10) + if (interval.TotalMinutes < 1) { - throw new ArgumentException("The minimum interval must be at least 10 minutes."); + throw new ArgumentException("The minimum interval must be at least 1 minutes."); } return nextOccurrence - interval; } diff --git a/src/Infrastructure/BotSharp.Core.Rules/Engines/IRuleEngine.cs b/src/Infrastructure/BotSharp.Core.Rules/Engines/IRuleEngine.cs index f7db3a96..45eae430 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/Engines/IRuleEngine.cs +++ b/src/Infrastructure/BotSharp.Core.Rules/Engines/IRuleEngine.cs @@ -1,5 +1,3 @@ -using BotSharp.Core.Rules.Triggers; - namespace BotSharp.Core.Rules.Engines; public interface IRuleEngine diff --git a/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs b/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs index 4fdf66ba..feea122f 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs +++ b/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs @@ -1,10 +1,7 @@ using BotSharp.Abstraction.Conversations; -using BotSharp.Abstraction.Conversations.Enums; using BotSharp.Abstraction.Models; using BotSharp.Abstraction.Repositories.Filters; -using BotSharp.Abstraction.Routing; using BotSharp.Abstraction.Utilities; -using BotSharp.Core.Rules.Triggers; using Microsoft.Extensions.Logging; using System.Data; diff --git a/src/Infrastructure/BotSharp.Core.Rules/Using.cs b/src/Infrastructure/BotSharp.Core.Rules/Using.cs index 5a83b419..a4353c96 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/Using.cs +++ b/src/Infrastructure/BotSharp.Core.Rules/Using.cs @@ -6,4 +6,6 @@ global using BotSharp.Abstraction.Plugins; global using BotSharp.Abstraction.Agents; global using BotSharp.Abstraction.Conversations.Models; global using BotSharp.Abstraction.Instructs; -global using BotSharp.Abstraction.Instructs.Models; \ No newline at end of file +global using BotSharp.Abstraction.Instructs.Models; + +global using BotSharp.Abstraction.Rules; \ No newline at end of file diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs index cee5cc6f..5c49f328 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs @@ -33,7 +33,7 @@ public class RedisPublisher : IEventPublisher if (CheckMessageExists(db, channel, "message", message)) { - _logger.LogError($"The message already exists {channel} {message}"); + _logger.LogInformation($"The message already exists {channel} {message}"); return null; } @@ -84,7 +84,8 @@ public class RedisPublisher : IEventPublisher return [ new NameValueEntry("message", message), - new NameValueEntry("timestamp", DateTime.UtcNow.ToString("o")) + new NameValueEntry("timestamp", DateTime.UtcNow.ToString("o")), + new NameValueEntry("machine", Environment.MachineName) ]; } @@ -94,6 +95,7 @@ public class RedisPublisher : IEventPublisher [ new NameValueEntry("message", message), new NameValueEntry("timestamp", DateTime.UtcNow.ToString("o")), + new NameValueEntry("machine", Environment.MachineName), new NameValueEntry("error", error) ]; } diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/RulesController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/RulesController.cs index 613f82c2..f135866f 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/RulesController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/RulesController.cs @@ -1,4 +1,5 @@ using BotSharp.Abstraction.Agents.Models; +using BotSharp.Abstraction.Rules; using BotSharp.Core.Rules.Triggers; namespace BotSharp.OpenAPI.Controllers; diff --git a/src/Plugins/BotSharp.Plugin.HttpHandler/Functions/HandleHttpRequestFn.cs b/src/Plugins/BotSharp.Plugin.HttpHandler/Functions/HandleHttpRequestFn.cs index 607ae5a9..f198c47c 100644 --- a/src/Plugins/BotSharp.Plugin.HttpHandler/Functions/HandleHttpRequestFn.cs +++ b/src/Plugins/BotSharp.Plugin.HttpHandler/Functions/HandleHttpRequestFn.cs @@ -188,6 +188,11 @@ public class HandleHttpRequestFn : IFunctionCallback { if (response == null) return string.Empty; + if (response.StatusCode != System.Net.HttpStatusCode.OK) + { + return response.ReasonPhrase ?? "http call has error occurred."; + } + return await response.Content.ReadAsStringAsync(); } }