build-flow.api/src/BuildFlow.Host/Configuration/ConsulConfigurationProvider.cs
2026-07-23 18:15:40 +08:00

174 lines
6.4 KiB
C#
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

using System.Net.Http.Headers;
namespace BuildFlow.Host.Configuration;
/// <summary>
/// 轻量 Consul KV 配置 Provider(自实现,替代 Winton.Extensions.Configuration.Consul)。
/// 启动时用 Consul HTTP API 拉取 prefix 下所有 KV,扁平化注入 IConfiguration。
/// 支持 ReloadOnChange(后台 long-poll watch)。
///
/// 与 Winton 的区别:完全可控、有日志、不黑盒;用 HttpClient 直接调 /v1/kv/?recurse=true。
///
/// KV key 映射:Consul key 去掉 root 前缀后,剩余部分直接作为 IConfiguration key。
/// 如 Consul key "buildflow/Jwt:SigningKey",root="buildflow/",剩余 "Jwt:SigningKey" → IConfiguration key "Jwt:SigningKey"。
/// </summary>
internal sealed class ConsulConfigurationSource : IConfigurationSource
{
private readonly ConsulOptions _options;
public ConsulConfigurationSource(ConsulOptions options) => _options = options;
public IConfigurationProvider Build(IConfigurationBuilder builder) =>
new ConsulConfigurationProvider(_options);
}
internal sealed class ConsulOptions
{
public required string Host { get; init; } // 如 http://43.134.235.58:8520
public required string Token { get; init; } // ACL token
public required string KeyPrefix { get; init; } // 如 buildflow/
public bool Optional { get; init; } = true;
public bool ReloadOnChange { get; init; } = true;
public TimeSpan PollInterval { get; init; } = TimeSpan.FromSeconds(30);
}
internal sealed class ConsulConfigurationProvider : ConfigurationProvider, IDisposable
{
private readonly ConsulOptions _options;
private readonly HttpClient _http;
private readonly ILoggerFactory? _loggerFactory;
private readonly ILogger<ConsulConfigurationProvider> _logger;
private ulong _lastIndex;
private Timer? _watchTimer;
private bool _disposed;
public ConsulConfigurationProvider(ConsulOptions options)
{
_options = options;
_http = new HttpClient { Timeout = TimeSpan.FromSeconds(10) };
if (!string.IsNullOrEmpty(options.Token))
_http.DefaultRequestHeaders.Add("X-Consul-Token", options.Token);
_loggerFactory = LoggerFactory.Create(b => b.AddConsole());
_logger = _loggerFactory.CreateLogger<ConsulConfigurationProvider>();
}
public override void Load()
{
try
{
var (dict, index) = FetchKvAsync(waitForChange: false).GetAwaiter().GetResult();
_lastIndex = index;
Data = dict;
_logger.LogInformation("[Consul] 加载成功:{Count} 个 KV,LastIndex={Index}", dict.Count, index);
if (_options.ReloadOnChange && _watchTimer == null)
{
_watchTimer = new Timer(_ => WatchCallback(), null, _options.PollInterval, _options.PollInterval);
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "[Consul] 加载失败(Optional={Optional}): {Message}", _options.Optional, ex.Message);
if (!_options.Optional) throw;
}
}
private async void WatchCallback()
{
if (_disposed) return;
try
{
var (dict, index) = await FetchKvAsync(waitForChange: true);
if (index > _lastIndex)
{
_lastIndex = index;
Data = dict;
OnReload();
_logger.LogInformation("[Consul] 热重载:检测到变更(LastIndex={Index}),已刷新 {Count} 个 KV", index, dict.Count);
}
}
catch (Exception ex)
{
_logger.LogDebug(ex, "[Consul] watch 轮询异常(将继续重试): {Message}", ex.Message);
}
}
private async Task<(Dictionary<string, string?> Dict, ulong Index)> FetchKvAsync(bool waitForChange)
{
// Consul KV recurse API:返回 prefix 下所有 KV
var url = $"{_options.Host}/v1/kv/{_options.KeyPrefix}?recurse=true";
using var resp = await _http.GetAsync(url);
if (resp.StatusCode == System.Net.HttpStatusCode.NotFound)
{
// Consul 返回 404:无此 key(首次未灌入)。Optional 时视为空配置。
var idx = ParseIndex(resp.Headers);
return (new Dictionary<string, string?>(), idx);
}
resp.EnsureSuccessStatusCode();
var kvs = await resp.Content.ReadFromJsonAsync<List<ConsulKvItem>>()
?? new List<ConsulKvItem>();
var lastIndex = ParseIndex(resp.Headers);
var dict = new Dictionary<string, string?>(StringComparer.OrdinalIgnoreCase);
foreach (var kv in kvs)
{
// key 去掉 prefix,剩余直接作为 IConfiguration key(Consul key 里用 : 表层级)
var configKey = kv.Key;
if (configKey.StartsWith(_options.KeyPrefix, StringComparison.OrdinalIgnoreCase))
configKey = configKey[_options.KeyPrefix.Length..];
// Consul key 可能用 / 分隔层级,转成 IConfiguration 的 :
configKey = configKey.Replace('/', ':');
// value 是 base64 编码的,Consul API 默认返回 base64
var value = DecodeValue(kv.Value);
dict[configKey] = value;
}
return (dict, lastIndex);
}
private static ulong ParseIndex(HttpResponseHeaders headers)
{
if (headers.TryGetValues("X-Consul-Index", out var vals))
{
var s = vals.FirstOrDefault();
if (ulong.TryParse(s, out var idx)) return idx;
}
return 0;
}
// Consul KV API 的 Value 字段是 base64 编码的(recurse 接口默认)
private static string? DecodeValue(string? b64)
{
if (string.IsNullOrEmpty(b64)) return null;
try
{
var bytes = Convert.FromBase64String(b64);
return System.Text.Encoding.UTF8.GetString(bytes);
}
catch
{
return b64; // 非 base64,原样返回
}
}
public void Dispose()
{
_disposed = true;
_watchTimer?.Dispose();
_http.Dispose();
_loggerFactory?.Dispose();
}
}
// Consul KV API 响应项(recurse 接口)
internal sealed class ConsulKvItem
{
public int CreateIndex { get; set; }
public int ModifyIndex { get; set; }
public string Key { get; set; } = "";
public int LockIndex { get; set; }
public string? Value { get; set; }
public int Flags { get; set; }
}