diff --git a/src/BuildFlow.Host/BuildFlow.Host.csproj b/src/BuildFlow.Host/BuildFlow.Host.csproj index 3679945..23bcf6d 100644 --- a/src/BuildFlow.Host/BuildFlow.Host.csproj +++ b/src/BuildFlow.Host/BuildFlow.Host.csproj @@ -54,6 +54,8 @@ + + diff --git a/src/BuildFlow.Host/Program.cs b/src/BuildFlow.Host/Program.cs index 82d055e..0834533 100644 --- a/src/BuildFlow.Host/Program.cs +++ b/src/BuildFlow.Host/Program.cs @@ -1,6 +1,7 @@ using BuildFlow.Host; using BuildFlow.Host.Configuration; using OpenTelemetry.Logs; +using Winton.Extensions.Configuration.Consul; var builder = WebApplication.CreateBuilder(args); @@ -8,6 +9,21 @@ var builder = WebApplication.CreateBuilder(args); // 不存在时 optional:true 静默跳过，回退到 appsettings.json 中的占位符。 builder.Configuration.AddJsonFile("appsettings.Secrets.json", optional: true, reloadOnChange: true); +// Consul KV 配置中心（优先级最高，覆盖本地 json + secrets，作为生产配置真相源）。 +// 加载顺序决定优先级：后注册的覆盖先注册的，故 Consul 在 Secrets 之后。 +// 鸡生蛋问题：Consul 地址本身用环境变量 Consul__Host 注入（compose 配置）， +// 默认回退到 middleware 公网地址。Consul 不可用时 Optional=true 回退本地 json，不阻断启动。 +var consulHost = builder.Configuration["Consul:Host"] ?? "http://43.134.235.58:8500"; +builder.Configuration.AddConsul( + "buildflow/", + options => + { + options.ConsulConfigurationOptions = co => co.Address = new Uri(consulHost); + options.Optional = true; // Consul 不可用不阻断启动，回退本地 appsettings + options.ReloadOnChange = true; // KV 变更自动刷新 IConfiguration（热重载基础） + options.OnLoadException = ex => ex.Ignore = true; // 容错：Consul 抖动不影响应用 + }); + // OTel 日志导出 var otlpEndpoint = builder.Configuration["OpenTelemetry:OtlpEndpoint"] ?? "http://localhost:4316"; builder.Logging.AddOpenTelemetry(logging => diff --git a/src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs b/src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs index ae3029d..432f238 100644 --- a/src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs +++ b/src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs @@ -15,26 +15,28 @@ public class ImportJobProcessorService : BackgroundService { private readonly IServiceScopeFactory _scopeFactory; private readonly ILogger _logger; - private readonly NotificationOptions _options; + private readonly IOptionsMonitor _options; public ImportJobProcessorService( IServiceScopeFactory scopeFactory, ILogger logger, - IOptions options) + IOptionsMonitor options) { _scopeFactory = scopeFactory; _logger = logger; - _options = options.Value; + _options = options; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { - _logger.LogInformation("ImportJobProcessor 启动，轮询间隔 {Interval}s", _options.Scheduler.PollIntervalSeconds); + // timer 间隔启动时定一次（方案 A：timer 参数不热重载，改间隔需重启服务） + var opts = _options.CurrentValue; + _logger.LogInformation("ImportJobProcessor 启动，轮询间隔 {Interval}s", opts.Scheduler.PollIntervalSeconds); // 启动时先等待一轮，避免与应用启动并发争抢资源 - await Task.Delay(_options.Scheduler.StartupDelaySeconds * 1000, stoppingToken); + await Task.Delay(opts.Scheduler.StartupDelaySeconds * 1000, stoppingToken); - using var timer = new PeriodicTimer(TimeSpan.FromSeconds(_options.Scheduler.PollIntervalSeconds)); + using var timer = new PeriodicTimer(TimeSpan.FromSeconds(opts.Scheduler.PollIntervalSeconds)); // 整个 do-while 纳入 try/catch：timer.WaitForNextTickAsync 在 stoppingToken 取消时 // 会抛 OperationCanceledException，该异常发生在 while 条件求值处（不在内层 try 范围）， diff --git a/src/BuildFlow.Host/Services/BackgroundServices/MonitoringSyncSchedulerService.cs b/src/BuildFlow.Host/Services/BackgroundServices/MonitoringSyncSchedulerService.cs index 7c832df..62e1913 100644 --- a/src/BuildFlow.Host/Services/BackgroundServices/MonitoringSyncSchedulerService.cs +++ b/src/BuildFlow.Host/Services/BackgroundServices/MonitoringSyncSchedulerService.cs @@ -17,23 +17,23 @@ namespace BuildFlow.Host.Services.BackgroundServices; /// public class MonitoringSyncSchedulerService( IServiceScopeFactory scopeFactory, - IOptions options, + IOptionsMonitor options, ILogger logger) : BackgroundService { - private readonly MonitoringOptions _options = options.Value; - protected override async Task ExecuteAsync(CancellationToken stoppingToken) { + // timer 间隔启动时定一次（方案 A：timer 参数不热重载，改间隔需重启服务） + var opts = options.CurrentValue; // 整个方法体纳入 try/catch：Task.Delay / WaitForNextTickAsync 在进程关闭时 // 抛 OperationCanceledException，不捕获会触发 StopHost 崩溃整个进程。 try { - await Task.Delay(_options.Scheduler.StartupDelaySeconds * 1000, stoppingToken); + await Task.Delay(opts.Scheduler.StartupDelaySeconds * 1000, stoppingToken); logger.LogInformation("MonitoringSyncScheduler 启动，轮询间隔 {Interval}s", - _options.Scheduler.PollIntervalSeconds); + opts.Scheduler.PollIntervalSeconds); - using var timer = new PeriodicTimer(TimeSpan.FromSeconds(_options.Scheduler.PollIntervalSeconds)); + using var timer = new PeriodicTimer(TimeSpan.FromSeconds(opts.Scheduler.PollIntervalSeconds)); do { @@ -67,7 +67,8 @@ public class MonitoringSyncSchedulerService( var processor = scope.ServiceProvider.GetRequiredService(); var now = DateTime.UtcNow; - var defaultInterval = _options.Scheduler.PollIntervalSeconds; + // 每轮扫描读最新配置（业务参数热重载，timer 间隔除外） + var defaultInterval = options.CurrentValue.Scheduler.PollIntervalSeconds; // 启用/间隔/最近同步时间都在 json_data 内，先物化全部再在内存过滤 // （外部系统数量很少，全表扫描无压力；若未来量大再加 jsonb 索引或冗余列） diff --git a/src/BuildFlow.Host/Services/BackgroundServices/TimeoutSchedulerService.cs b/src/BuildFlow.Host/Services/BackgroundServices/TimeoutSchedulerService.cs index 85c91e2..4f53420 100644 --- a/src/BuildFlow.Host/Services/BackgroundServices/TimeoutSchedulerService.cs +++ b/src/BuildFlow.Host/Services/BackgroundServices/TimeoutSchedulerService.cs @@ -14,26 +14,28 @@ public class TimeoutSchedulerService : BackgroundService { private readonly IServiceScopeFactory _scopeFactory; private readonly ILogger _logger; - private readonly NotificationOptions _options; + private readonly IOptionsMonitor _options; public TimeoutSchedulerService( IServiceScopeFactory scopeFactory, ILogger logger, - IOptions options) + IOptionsMonitor options) { _scopeFactory = scopeFactory; _logger = logger; - _options = options.Value; + _options = options; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { - _logger.LogInformation("TimeoutScheduler 启动，轮询间隔 {Interval}s", _options.Scheduler.PollIntervalSeconds); + // timer 间隔启动时定一次（方案 A：timer 参数不热重载，改间隔需重启服务） + var opts = _options.CurrentValue; + _logger.LogInformation("TimeoutScheduler 启动，轮询间隔 {Interval}s", opts.Scheduler.PollIntervalSeconds); // 启动时先等待一轮，避免与应用启动并发争抢资源 - await Task.Delay(_options.Scheduler.StartupDelaySeconds * 1000, stoppingToken); + await Task.Delay(opts.Scheduler.StartupDelaySeconds * 1000, stoppingToken); - using var timer = new PeriodicTimer(TimeSpan.FromSeconds(_options.Scheduler.PollIntervalSeconds)); + using var timer = new PeriodicTimer(TimeSpan.FromSeconds(opts.Scheduler.PollIntervalSeconds)); // 整个 do-while 纳入 try/catch：WaitForNextTickAsync 取消时抛的 OperationCanceledException // 发生在 while 条件处（不在内层 try），不捕获会触发 StopHost 崩溃进程。 diff --git a/src/BuildFlow.Host/Services/BackgroundServices/WebhookDispatcherService.cs b/src/BuildFlow.Host/Services/BackgroundServices/WebhookDispatcherService.cs index c09a8c8..dee99b7 100644 --- a/src/BuildFlow.Host/Services/BackgroundServices/WebhookDispatcherService.cs +++ b/src/BuildFlow.Host/Services/BackgroundServices/WebhookDispatcherService.cs @@ -21,34 +21,36 @@ public class WebhookDispatcherService : BackgroundService private readonly IServiceScopeFactory _scopeFactory; private readonly IHttpClientFactory _httpClientFactory; private readonly ILogger _logger; - private readonly NotificationOptions _options; + private readonly IOptionsMonitor _options; public WebhookDispatcherService( IServiceScopeFactory scopeFactory, IHttpClientFactory httpClientFactory, - IOptions options, + IOptionsMonitor options, ILogger logger) { _scopeFactory = scopeFactory; _httpClientFactory = httpClientFactory; _logger = logger; - _options = options.Value; + _options = options; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { + // 启动时读一次：AllowedHosts 决定是否启用、timer 间隔（方案 A：不热重载） + var opts = _options.CurrentValue; // 未配置任何 AllowedHosts 时，Webhook 投递被完全禁用，服务空转退出（避免无谓轮询） - if (_options.Webhook.AllowedHosts.Length == 0) + if (opts.Webhook.AllowedHosts.Length == 0) { _logger.LogInformation("WebhookDispatcher：AllowedHosts 为空，Webhook 投递已禁用，服务退出"); return; } - _logger.LogInformation("WebhookDispatcher 启动，轮询间隔 {Interval}s", _options.Webhook.PollIntervalSeconds); + _logger.LogInformation("WebhookDispatcher 启动，轮询间隔 {Interval}s", opts.Webhook.PollIntervalSeconds); - await Task.Delay(_options.Scheduler.StartupDelaySeconds * 1000, stoppingToken); + await Task.Delay(opts.Scheduler.StartupDelaySeconds * 1000, stoppingToken); - using var timer = new PeriodicTimer(TimeSpan.FromSeconds(_options.Webhook.PollIntervalSeconds)); + using var timer = new PeriodicTimer(TimeSpan.FromSeconds(opts.Webhook.PollIntervalSeconds)); do { @@ -89,7 +91,7 @@ public class WebhookDispatcherService : BackgroundService return; var client = _httpClientFactory.CreateClient("Webhook"); - client.Timeout = TimeSpan.FromSeconds(_options.Webhook.TimeoutSeconds); + client.Timeout = TimeSpan.FromSeconds(_options.CurrentValue.Webhook.TimeoutSeconds); foreach (var delivery in pending) { @@ -186,7 +188,9 @@ public class WebhookDispatcherService : BackgroundService /// SSRF 防护：校验 URL 主机在白名单内，且拒绝 localhost 与私有/内网 IP。 internal bool IsHostAllowed(string url) { - if (_options.Webhook.AllowedHosts is null || _options.Webhook.AllowedHosts.Length == 0) + // 每次校验读最新配置（AllowedHosts 热重载，Consul KV 改白名单即时生效） + var allowedHosts = _options.CurrentValue.Webhook.AllowedHosts; + if (allowedHosts is null || allowedHosts.Length == 0) return false; if (!Uri.TryCreate(url, UriKind.Absolute, out var uri)) @@ -213,7 +217,7 @@ public class WebhookDispatcherService : BackgroundService } } - return _options.Webhook.AllowedHosts + return allowedHosts .Any(allowed => string.Equals(allowed, host, StringComparison.OrdinalIgnoreCase)); } } \ No newline at end of file diff --git a/src/modules/file/Application/Features/Files/Handlers/GetOfficePreviewQueryHandler.cs b/src/modules/file/Application/Features/Files/Handlers/GetOfficePreviewQueryHandler.cs index 314df88..980a1ed 100644 --- a/src/modules/file/Application/Features/Files/Handlers/GetOfficePreviewQueryHandler.cs +++ b/src/modules/file/Application/Features/Files/Handlers/GetOfficePreviewQueryHandler.cs @@ -15,15 +15,17 @@ namespace File.Application.Features.Files.Handlers; /// internalToken 为后端签发的短时 HMAC 签名（非用户 JWT），仅用于 KKFileView 内部回拉，10 分钟有效。 /// public class GetOfficePreviewQueryHandler( - IOptions kkOptions) : IRequestHandler + IOptionsMonitor kkOptions) : IRequestHandler { /// 内部回拉 token 有效期：KKFileView 拉取文件的时间窗口。 private const int TokenValidityMinutes = 10; public Task Handle(GetOfficePreviewQuery req, CancellationToken ct) { - var backendBase = kkOptions.Value.BackendBaseUrl.TrimEnd('/'); - var internalKey = kkOptions.Value.InternalKey; + // IOptionsMonitor.CurrentValue：每次请求读最新配置，Consul KV 改动实时生效 + var opts = kkOptions.CurrentValue; + var backendBase = opts.BackendBaseUrl.TrimEnd('/'); + var internalKey = opts.InternalKey; // 生成短时 HMAC 签名：objectKey + 过期时间戳 var expires = DateTimeOffset.UtcNow.AddMinutes(TokenValidityMinutes).ToUnixTimeSeconds(); @@ -37,7 +39,7 @@ public class GetOfficePreviewQueryHandler( // KKFileView onlinePreview 约定：url 参数为 Base64 编码的文件直链 var encoded = Convert.ToBase64String(Encoding.UTF8.GetBytes(proxyUrl)); - var kkBase = kkOptions.Value.BaseUrl.TrimEnd('/'); + var kkBase = opts.BaseUrl.TrimEnd('/'); var previewUrl = $"{kkBase}/onlinePreview?url={encoded}"; return Task.FromResult(new OfficePreviewResult(previewUrl)); diff --git a/src/modules/file/Infrastructure/FileInfrastructureModule.cs b/src/modules/file/Infrastructure/FileInfrastructureModule.cs index aa4b35a..609a0da 100644 --- a/src/modules/file/Infrastructure/FileInfrastructureModule.cs +++ b/src/modules/file/Infrastructure/FileInfrastructureModule.cs @@ -34,12 +34,10 @@ public class FileInfrastructureModule : AbpModule services.AddScoped(); // S3 兼容存储（单例，AmazonS3Client 线程安全可复用） + // 注入 IOptionsMonitor：Consul KV 改 S3 配置时 OnChange 回调重建 client，热重载生效 services.Configure(config.GetSection(S3Options.SectionName)); services.AddSingleton(sp => - { - var options = sp.GetRequiredService>().Value; - return new S3StorageService(Options.Create(options)); - }); + new S3StorageService(sp.GetRequiredService>())); // KKFileView 在线预览服务配置（Office/PDF 文档预览，生成预览地址时读取） services.Configure(config.GetSection(KkFileViewOptions.SectionName)); diff --git a/src/modules/file/Infrastructure/Storage/S3StorageService.cs b/src/modules/file/Infrastructure/Storage/S3StorageService.cs index 39cd276..32f0d09 100644 --- a/src/modules/file/Infrastructure/Storage/S3StorageService.cs +++ b/src/modules/file/Infrastructure/Storage/S3StorageService.cs @@ -12,11 +12,30 @@ namespace File.Infrastructure.Storage; /// public class S3StorageService : IS3StorageService { - private readonly IAmazonS3 _client; + // volatile：OnChange 回调会在配置变更时重建 client 并重新赋值该字段， + // volatile 保证其他线程立即看到新引用（避免读到半构造对象）。 + private volatile IAmazonS3 _client; + private readonly Lock _clientLock = new(); - public S3StorageService(IOptions options) + public S3StorageService(IOptionsMonitor options) + { + _client = BuildClient(options.CurrentValue); + // 配置变更（如 Consul KV 改 Endpoint/凭据）时重建 client，热重载生效。 + // 重建期间并发的旧请求仍用旧 client（volatile 读），重建完成后新请求用新 client。 + options.OnChange(opt => + { + var newClient = BuildClient(opt); + lock (_clientLock) + { + var old = _client; + _client = newClient; + old?.Dispose(); + } + }); + } + + private static AmazonS3Client BuildClient(S3Options opt) { - var opt = options.Value; var region = string.IsNullOrEmpty(opt.Region) ? "us-east-1" : opt.Region; var config = new AmazonS3Config { @@ -25,7 +44,7 @@ public class S3StorageService : IS3StorageService // ForcePathStyle：RustFS/MinIO 等 S3 兼容存储要求 path-style 寻址（bucket 放在路径而非子域）。 ForcePathStyle = true, }; - _client = new AmazonS3Client(opt.AccessKey, opt.SecretKey, config); + return new AmazonS3Client(opt.AccessKey, opt.SecretKey, config); } /// 把 S3 的 NotFound 类错误包装为领域 NotFoundException，其他原样抛出。 diff --git a/src/modules/rag/Infrastructure/AI/AiModelRuntimeConfigResolver.cs b/src/modules/rag/Infrastructure/AI/AiModelRuntimeConfigResolver.cs index 1f59ada..18c6840 100644 --- a/src/modules/rag/Infrastructure/AI/AiModelRuntimeConfigResolver.cs +++ b/src/modules/rag/Infrastructure/AI/AiModelRuntimeConfigResolver.cs @@ -8,18 +8,19 @@ namespace BuildFlow.Infrastructure.AI; /// /// AI 运行时模型配置解析器。优先读取模型管理中的默认启用模型，没有配置时回退到 appsettings。 +/// 配置通过 IOptionsMonitor 实时读取，Consul KV 改动即时生效（无需重启）。 /// public class AiModelRuntimeConfigResolver( IServiceScopeFactory scopeFactory, - IOptions options) + IOptionsMonitor options) { - private readonly AiOptions _fallbackOptions = options.Value; - /// /// 按模型类型解析本次调用配置。Chat 与 Embedding 都通过该入口，避免某一类调用绕过本地模型配置。 /// public async Task ResolveAsync(ModelType modelType, CancellationToken ct) { + // 每次解析读最新配置（IOptionsMonitor.CurrentValue），支持热重载 + var fallback = options.CurrentValue; using var scope = scopeFactory.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); var model = await db.AiModels @@ -32,31 +33,31 @@ public class AiModelRuntimeConfigResolver( model.ModelId, model.BaseUrl, model.ApiKey, - ResolveDefaultInstructions(model.DefaultInstructions), + ResolveDefaultInstructions(model.DefaultInstructions, fallback), model.MaxTokens, model.Temperature); } var fallbackModelId = modelType == ModelType.Embedding - ? _fallbackOptions.EmbeddingModel - : _fallbackOptions.ChatModel; + ? fallback.EmbeddingModel + : fallback.ChatModel; return new AiModelRuntimeConfig( fallbackModelId, - _fallbackOptions.BaseUrl, - _fallbackOptions.ApiKey, - _fallbackOptions.DefaultInstructions, - _fallbackOptions.MaxTokens, - _fallbackOptions.Temperature); + fallback.BaseUrl, + fallback.ApiKey, + fallback.DefaultInstructions, + fallback.MaxTokens, + fallback.Temperature); } - private string? ResolveDefaultInstructions(string? modelInstructions) + private static string? ResolveDefaultInstructions(string? modelInstructions, AiOptions fallback) { if (string.IsNullOrWhiteSpace(modelInstructions)) - return _fallbackOptions.DefaultInstructions; + return fallback.DefaultInstructions; return IsLegacyRagOnlyInstructions(modelInstructions) - ? _fallbackOptions.DefaultInstructions + ? fallback.DefaultInstructions : modelInstructions; } diff --git a/src/modules/workflow/Application/Notifications/NotificationService.cs b/src/modules/workflow/Application/Notifications/NotificationService.cs index 4b77522..1478b7c 100644 --- a/src/modules/workflow/Application/Notifications/NotificationService.cs +++ b/src/modules/workflow/Application/Notifications/NotificationService.cs @@ -17,10 +17,10 @@ namespace Workflow.Application.Notifications; /// public class NotificationService( WorkflowDbContext db, - IOptions options, + IOptionsMonitor options, IPlatformNotificationPublisher platformPublisher) : INotificationService { - private readonly NotificationOptions _options = options.Value; + // 配置通过 IOptionsMonitor 实时读取（Scoped，每请求重新 resolve），Consul KV 改动即时生效 public Task NotifyTaskArrivedAsync(WorkflowTask task, CancellationToken ct = default) => CreateAndDispatchAsync(task, "task-arrived", @@ -179,7 +179,7 @@ public class NotificationService( HttpMethod = "POST", Status = "pending", Attempts = 0, - MaxAttempts = _options.Webhook.MaxAttempts, + MaxAttempts = options.CurrentValue.Webhook.MaxAttempts, NextRetryAt = now, Category = category, RelatedInstanceId = task.InstanceId, @@ -197,7 +197,7 @@ public class NotificationService( private string ResolveWebhookUrl() { // 当前实现使用全局 DefaultUrl；节点级 webhookUrl 配置可后续扩展（需 ProcessEngine 传入 node.Config） - var url = _options.Webhook.DefaultUrl; + var url = options.CurrentValue.Webhook.DefaultUrl; return string.IsNullOrWhiteSpace(url) ? string.Empty : url; } @@ -207,14 +207,15 @@ public class NotificationService( /// internal bool IsHostAllowed(string url) { - if (_options.Webhook.AllowedHosts is null || _options.Webhook.AllowedHosts.Length == 0) + var webhook = options.CurrentValue.Webhook; + if (webhook.AllowedHosts is null || webhook.AllowedHosts.Length == 0) return false; if (!Uri.TryCreate(url, UriKind.Absolute, out var uri)) return false; var host = uri.Host; - return _options.Webhook.AllowedHosts + return webhook.AllowedHosts .Any(allowed => string.Equals(allowed, host, StringComparison.OrdinalIgnoreCase)); } }