Elsa Structured Log Persistence：为 Elsa Workflow Engine 构建可插拔的日志持久化存储
Elsa Structured Log Persistence：为 Elsa Workflow Engine 构建可插拔的日志持久化存储
结构化日志（Structured Log）是 Elsa 诊断体系中的核心能力：它捕获 ILogger 语义日志、脱敏敏感值、维护有界的内存近端缓冲，并通过 REST 端点与 SignalR Hub 将实时日志流推送到 Studio。本篇文章基于仓库 specs/005-structured-log-persistence/plan.md 展开，系统讲解该特性的实现方案：如何在保留现有 IStructuredLogProvider 作为 REST/SignalR 门面的前提下，将可查询存储重构为可替换的抽象，并以 SQLite 作为首个可选持久化提供者。读完本文，你将掌握 Elsa 结构化日志存储的分层架构、FluentMigrator 模式管理、有界写队列与保留策略的设计取舍，并能直接上手启用内存存储与 SQLite 持久化两种模式。
特性概览：从"有界内存"到"可选持久化"
结构化日志持久化特性（feature 005-structured-log-persistence）的核心目标可以概括为一句话：把可查询的日志存储变成可插拔的关切点，同时保持现有模块行为不回归。原始需求描述为："Make structured log storage a pluggable concern. Keep in-memory support, add an easy durable SQLite storage story first, use FluentMigrator for schema management, and defer OpenTelemetry/exporter scope."
按 spec.md 的澄清记录，该特性做了以下关键范围界定：
- 不引入任何厂商 sink：Logstash、Datadog、Splunk、Loki、Seq 等导出器全部推迟；
- 不引入 OTLP 日志导出器：OpenTelemetry 边界留给后续 diagnostics OpenTelemetry 特性；
- 首版存储提供者：保留内存存储（默认），新增 SQLite 持久化（可选）；
- SQLite 不是一次性实现：它是第一个关系型提供者，为后续 SQL Server、PostgreSQL、MySQL 等提供者可复用的关系持久化模型；
- 不使用 EF Core：本特性避免 EF Core 及逐提供者的 EF 迁移，模式管理与热路径写入改用 FluentMigrator + 显式 SQL/Dapper 风格访问。
从仓库现状看，该计划已经完整落地：核心模块 src/modules/Elsa.Diagnostics.StructuredLogs/、共享关系持久化包 src/modules/Elsa.Diagnostics.StructuredLogs.Persistence.Relational/、SQLite 提供者包 src/modules/Elsa.Diagnostics.StructuredLogs.Persistence.Sqlite/ 均已存在，并配套了关系型单元测试与 SQLite 集成测试项目。
技术背景与关键依赖
plan.md 明确了本特性的技术栈：
- 语言/版本：最新 C#，启用可空引用类型（nullable reference types）与隐式 using；
- 主要依赖：现有
Elsa.Diagnostics.StructuredLogs模块、Microsoft.Extensions.Logging、Microsoft.Extensions.Options、Microsoft.AspNetCore.SignalR、Elsa 特性/模块基础设施、FastEndpoints（通过 Elsa API 端点模式）、FluentMigrator runner 包、SQLite ADO.NET 提供者，关系操作可选 Dapper； - 存储形态：默认有界内存存储；可选 SQLite 持久化（基于共享关系持久化）。SQLite 将
Timestamp与ReceivedAt存储为 UTC ISO-8601 文本，异常、作用域与属性载荷存储为 JSON 文本； - 目标平台：仓库支持的 .NET 目标框架上的 ASP.NET Core Elsa Server；
- 项目形态：现有 Elsa 解决方案内的模块化 .NET 库。
架构设计：四大边界与一个门面
核心原则：拆分职责、保留门面
方案的首要约束来自需求 FR-001/FR-002：必须将可查询存储与实时流、捕获/提供者门面职责分离；IStructuredLogProvider 继续充当 REST 端点与 SignalR 使用的门面，但存储专属职责必须由可替换的 store 抽象表达。仓库中 src/modules/Elsa.Diagnostics.StructuredLogs/Contracts/ 目录下的契约正是这一设计的落地：
IStructuredLogSink：只追加（append-only）的日志事件目的地，接收单个或批量事件，不必支持查询，供 store 内部及未来的 exporter 包使用；IStructuredLogStore（继承IStructuredLogSink）：可查询存储边界，追加脱敏后的StructuredLogEvent、按StructuredLogFilter查询近期事件、列出已知StructuredLogSource；IStructuredLogLiveFeed：实时投递边界，向订阅者发布新事件、应用订阅过滤器、在订阅者队列溢出时报告丢弃事件摘要，不承诺持久性；IStructuredLogProvider：现有门面契约，继续被 REST 端点与 SignalR 使用。
data-model.md 对各边界职责做了精炼定义：store "does not own live subscriber backpressure behavior"（不负责订阅者背压）；live feed "does not guarantee durability"（不承诺持久性）。这种单职责划分使得内存实现与 SQLite 实现可以各自独立演化。
项目结构决策
按 plan.md 的 "Structure Decision"：
将运行时捕获、API、Hub 及 Studio 侧契约保留在现有结构化日志模块中；只为持久化新增关系型与 SQLite 包。本特性不创建 OTLP 或厂商 sink 包。
仓库中的实际布局如下：
src/modules/
├── Elsa.Diagnostics.StructuredLogs/ # 运行时捕获 + REST + SignalR + 内存默认实现
│ ├── Contracts/ # IStructuredLogProvider/Store/LiveFeed/Sink/Redactor...
│ ├── Providers/InMemory/ # 有界内存 store/live feed + RingBuffer
│ ├── Services/DefaultStructuredLogProvider.cs
│ ├── Endpoints/StructuredLogs/{Recent,Sources,Storage}/
│ └── RealTime/StructuredLogsHub.cs
├── Elsa.Diagnostics.StructuredLogs.Persistence.Relational/ # 共享关系持久化
│ ├── Contracts/ Features/ Migrations/ Models/ Options/ Services/ Stores/
└── Elsa.Diagnostics.StructuredLogs.Persistence.Sqlite/ # SQLite 提供者
├── Extensions/ Features/ Options/ Services/ ShellFeatures/
测试侧对应 test/unit/Elsa.Diagnostics.StructuredLogs.UnitTests/、test/unit/Elsa.Diagnostics.StructuredLogs.Persistence.Relational.UnitTests/ 与 test/integration/Elsa.Diagnostics.StructuredLogs.Persistence.Sqlite.IntegrationTests/。
默认内存存储：零配置的有界历史
为什么保留内存默认
需求 FR-006 明确：未配置任何持久化提供者时，模块必须保持有界内存存储作为默认。这既是零配置的最快验证路径，也保证存储重构不回归现有行为（FR-007）。内存实现位于 Providers/InMemory：InMemoryStructuredLogStore、InMemoryStructuredLogLiveFeed 与 RingBuffer（有界环形缓冲）。
启用方式（来自 quickstart.md）：
services.AddElsa(elsa =>
{
elsa.UseStructuredLogs(options =>
{
options.RecentLogCapacity = 5_000;
options.MaxRecentLogQuerySize = 1_000;
});
});
关键配置项
结合 StructuredLogsOptions.cs 源码，内存模式的核心配置如下：
| 配置项 | 默认值 | 说明 |
|---|---|---|
RecentLogCapacity |
5_000 |
内存环形缓冲容量，超过上限时最旧事件被挤出，溢出计数通过查询结果暴露 |
MaxRecentLogQuerySize |
1_000 |
所有 IStructuredLogStore 实现的默认 Take 与上限钳制值；负值按 0 处理，查询构造不会抛异常 |
SubscriberChannelCapacity |
1_000 |
SignalR 实时订阅者通道容量，溢出时产生丢弃事件摘要 |
SourceHeartbeatTimeout |
30s |
源心跳超时，超过后源状态标记为 Stale |
IncludeStructuredLogsInternalLogs |
false |
是否包含模块内部日志 |
SensitiveNames / SensitiveTextPatterns |
见源码 | 脱敏名称集合与文本正则模式，可扩展 |
脱敏默认名称覆盖 authorization、token、password、secret、api-key、cookie、connection-string 等；默认文本模式匹配 Bearer token、key=value 形式密钥与 AccountKey/SharedAccessKey 连接串片段。所有事件在进入内存存储、SQLite 存储或实时订阅者之前，都必须先经过 IStructuredLogRedactor（FR-005）。
查询契约
StructuredLogFilter（StructuredLogFilter.cs）承载 REST 近期日志端点使用的过滤字段：MinimumLevel/Levels、CategoryPrefix、Text、TenantId、WorkflowDefinitionId、WorkflowInstanceId、TraceId、SpanId、CorrelationId、SourceId、From/To 时间范围与 Take 条数。
模块 README.md 还记录了一个易踩的契约细节：查询顺序固定为 Timestamp, ReceivedAt, SourceId, Sequence, Id；QueryAsync.DroppedEvents 字段表示内存环形缓冲溢出计数，而关系型写队列的丢弃量通过存储诊断上报、不走该字段；仅大小写不同的等值过滤不可移植——内存实现是 OrdinalIgnoreCase，SQL 则遵循数据库排序规则。
SQLite 持久化：第一个可选耐用存储
启用方式
按 quickstart.md，启用 SQLite 需要添加持久化包并配置数据库文件：
services.AddElsa(elsa =>
{
elsa.UseStructuredLogs(structuredLogs =>
{
structuredLogs.UseSqliteStorage("Data Source=elsa-structured-logs.db", sqlite =>
{
sqlite.RunMigrationsOnStartup = true;
sqlite.Relational.WriteQueue.Capacity = 10_000;
sqlite.Relational.WriteQueue.BatchSize = 100;
});
});
});
随后映射既有端点与 Hub：
app.UseStructuredLogs();
Studio 继续使用以下不变契约（FR-024/用户故事 3）：
/diagnostics/structured-logs/recent/diagnostics/structured-logs/sources/diagnostics/structured-logs/storage/elsa/hubs/diagnostics/structured-logs
注册链源码解析
SqliteStructuredLogsModuleExtensions.cs 展示了注册的内部机制：
UseSqliteStorage通过feature.Module.Use<SqliteStructuredLogPersistenceFeature>(...)挂载 SQLite 特性，并注入连接字符串与配置回调；AddSqliteStructuredLogPersistence注册选项（SqliteStructuredLogOptions与拷贝自 SQLite 的RelationalStructuredLogOptions），并TryAddSingleton注册连接工厂、方言、模式迁移器与启动服务；- 启动服务同时注册为
IHostedService与IStartupTask，保证在 Elsa 启动任务管线中执行； - 最后调用
AddRelationalStructuredLogPersistence()挂载共享关系持久化。
配置项与默认值
SqliteStructuredLogOptions.cs：
ConnectionString：默认Data Source=elsa-structured-logs.db；RunMigrationsOnStartup：默认true，启动自动迁移，可关闭；Relational：内嵌关系型配置（写队列 + 保留策略）。
RelationalStructuredLogOptions.cs：
| 配置分组 | 参数 | 默认值 | 说明 |
|---|---|---|---|
WriteQueue.Capacity |
队列容量 | 10_000 |
最大排队等待写入的事件数 |
WriteQueue.BatchSize |
批量大小 | 100 |
单次刷盘最多写入事件数 |
WriteQueue.FlushInterval |
刷盘间隔 | 1s |
后台两次刷盘之间的最大间隔 |
WriteQueue.ShutdownFlushTimeout |
关闭刷盘超时 | 10s |
优雅关闭时刷盘的最长耗时 |
Retention.MaxAge |
最大事件年龄 | null |
配置后清理超龄记录 |
Retention.MaxRows |
最大保留行数 | null |
配置后清理超额记录 |
Retention.CleanupOnStartup |
启动即清理 | false |
迁移后立即执行清理 |
关系型记录模型与模式管理
RelationalStructuredLogRecord.cs 完整映射了 StructuredLogEvent：
- 标量可查询列：
Id、Sequence（日志路径分配的单调序号）、Timestamp/ReceivedAt（UTC ISO-8601 文本）、Level、Category、EventId、EventName、Message、MessageTemplate（来自{OriginalFormat}）、TraceId、SpanId、CorrelationId、TenantId、WorkflowDefinitionId、WorkflowInstanceId、SourceId； - JSON 载荷列：
ExceptionJson（序列化的StructuredLogException）、ScopesJson、PropertiesJson（作用域键值对与结构化属性键值对）。
首版模式迁移 M001_CreateStructuredLogTables.cs（版本号 2026051301）创建 StructuredLogEvents 表，并为 Timestamp、ReceivedAt（降序索引）、Level、Category、SourceId、TenantId、WorkflowDefinitionId、WorkflowInstanceId、CorrelationId、TraceId 建立索引，配合 data-model.md 建议的 ReceivedAt, Sequence, Id 复合索引支撑近期排序查询。
迁移即启动与多实例考虑
SqliteStructuredLogStartupService.cs 实现启动逻辑：用 SemaphoreSlim 保证幂等执行；RunMigrationsOnStartup 为真时先跑 FluentMigrator 迁移，随后若 CleanupOnStartup 为真则执行保留清理。对 SQLite，启动迁移是推荐默认（quickstart 明确说明）。对于未来共享关系型数据库（SQL Server/PostgreSQL），生产环境可能选择在部署时只跑一次迁移而非每个实例都跑；多实例启动锁定的注意事项必须由各提供者文档化（FR-030）。
写缓冲与有界队列：热路径不阻塞 ILogger
设计取舍
需求 FR-014 至 FR-019 对热路径提出了严格约束：日志写入必须批量化/排队化，ILogger 调用不能同步等待磁盘 I/O；必须使用有界写队列；队列满时丢弃新事件而非阻塞或无限占用内存；丢弃计数通过 IStructuredLogWriteBuffer.DroppedWriteCount 暴露并输出警告摘要。
对应契约（persistence-contract.md）：
public interface IStructuredLogWriteBuffer : IAsyncDisposable
{
long DroppedWriteCount { get; }
ValueTask EnqueueAsync(StructuredLogEvent logEvent, CancellationToken cancellationToken = default);
ValueTask FlushAsync(CancellationToken cancellationToken = default);
}
实现机制
StructuredLogWriteBuffer.cs 是核心实现（同时实现 IStructuredLogStore、IStructuredLogWriteBuffer、IHostedService）：
- 入队：
WriteAsync在锁内检查_queue.Count >= Capacity，满则Interlocked.Increment(ref _droppedWriteCount)并立即返回（不阻塞），否则入队并通过SemaphoreSlim _signal唤醒后台任务； - 后台刷盘：
ProcessQueueAsync等待信号或FlushInterval超时，然后按BatchSize取出批次调用底层 store 批量写入；写入失败时把整批计入丢弃计数； - 优雅关闭：
StopAsync先取消后台任务，再以ShutdownFlushTimeout为超时上限执行FlushAsync；若超时，CountPendingWritesAsDropped将剩余排队事件计入丢弃量； - 崩溃风险：进程在事件进入后台队列但未刷盘前崩溃，队列中未刷盘事件可能丢失（FR-016）——这是有意的取舍，换取 ILogger 调用绝不阻塞。
保留策略：默认不删除任何记录
需求 FR-020 与 spec 的澄清记录一致：默认不删除任何记录，除非配置了 MaxAge、MaxRows 或两者。StructuredLogRetentionService 按 CleanupInterval 后台清理，CleanupOnStartup 决定是否在迁移后立即清理。quickstart 中的完整保留配置示例：
sqlite.Relational.Retention.MaxAge = TimeSpan.FromDays(14);
sqlite.Relational.Retention.MaxRows = 250_000;
sqlite.Relational.Retention.CleanupOnStartup = true;
关系型扩展性：为未来数据库留好接口
用户故事 3 要求 SQLite 证明关系持久化模型可行，而不是成为死胡同实现。为此，plan.md 与 persistence-contract.md 定义了三个"提供者拥有"的接口：
public interface IRelationalStructuredLogConnectionFactory
{
ValueTask<DbConnection> OpenConnectionAsync(CancellationToken cancellationToken = default);
}
public interface IRelationalStructuredLogDialect
{
string ProviderName { get; }
string QuoteIdentifier(string identifier);
string ApplyLimit(string sql, int take);
}
public interface IStructuredLogSchemaMigrator
{
ValueTask MigrateAsync(CancellationToken cancellationToken = default);
}
- 连接工厂：提供者自己负责创建连接（SQLite 实现为
SqliteStructuredLogConnectionFactory）； - 方言：封装标识符引用、参数前缀、LIMIT/TAKE 语法、时间戳存储与比较、自由文本谓词构造以及 FluentMigrator 数据库分支所需的
ProviderName；未来可按实现需要补充时间戳、自由文本过滤或 JSON 处理的差异方法； - 模式迁移器：基于 FluentMigrator 的 schema 创建与升级服务。
数据模型层面，data-model.md 还定义了可选的 RelationalStructuredLogSourceRecord（源标识、进程/机器元数据、Kubernetes/容器元数据、生命周期时间戳与状态），供需要单独持久化源的 store 使用。未来提供者只需自备连接工厂、方言与 FluentMigrator runner 注册，即可复用共享的 RelationalStructuredLogStore 实现，且不改变 REST、SignalR 与 Studio 契约（FR-024）。
验证路径：测试与手动检查
自动化验证
quickstart 给出实施后的定向构建与测试命令：
dotnet build src/modules/Elsa.Diagnostics.StructuredLogs/Elsa.Diagnostics.StructuredLogs.csproj
dotnet build src/modules/Elsa.Diagnostics.StructuredLogs.Persistence.Sqlite/Elsa.Diagnostics.StructuredLogs.Persistence.Sqlite.csproj
dotnet test test/unit/Elsa.Diagnostics.StructuredLogs.UnitTests/Elsa.Diagnostics.StructuredLogs.UnitTests.csproj
dotnet test test/integration/Elsa.Diagnostics.StructuredLogs.Persistence.Sqlite.IntegrationTests/Elsa.Diagnostics.StructuredLogs.Persistence.Sqlite.IntegrationTests.csproj
仓库中 test/integration/Elsa.Diagnostics.StructuredLogs.Persistence.Sqlite.IntegrationTests/ 已包含与成功标准（SC-002 至 SC-010）一一对应的测试文件：
SqliteStructuredLogStoreTests.cs/SqliteStructuredLogFilterTests.cs：验证近期查询尊重 level、category、source、workflow、correlation、trace/span、时间范围与 limit 过滤；SqliteStructuredLogMigrationTests.cs：验证 FluentMigrator 从空库创建 schema、启动迁移默认执行且可关闭；SqliteStructuredLogRetentionTests.cs：验证仅在配置保留策略时删除过期/超额记录、默认不删除任何记录；SqliteStructuredLogWriteQueueTests.cs：验证队列满时丢弃新事件并报告丢弃计数、优雅关闭刷盘排队写入；SqliteStructuredLogTimestampTests.cs：验证 UTC ISO-8601 时间戳文本的存储与过滤一致性；SqliteStructuredLogSourceTests.cs/SqliteStructuredLogTestHost.cs：验证源列表与测试宿主搭建。
手动验证清单
- 以启用 SQLite 结构化日志存储的方式启动 Elsa Server；
- 用不同的 level、category、workflow ID 与 correlation ID 发出多条
ILogger记录； - 在 Studio 查询近期日志并验证过滤器生效；
- 使用同一个 SQLite 数据库文件重启宿主；
- 再次查询近期日志，确认重启前的数据仍然可查；
- 验证时间戳以 UTC ISO-8601 值存储与过滤；
- 在测试环境调低保留设置，确认清理删除了旧行或超额行；
- 在测试环境灌满写队列，确认新事件被丢弃且丢弃计数可见。
边界与后续范围
本特性明确把以下内容排除在范围之外（FR-033 与 quickstart "Out of scope" 章节）：OTLP、Logstash、Datadog、Splunk、Loki、Seq、Parquet 等导出器。这些可在持久化稳定后以 sink/exporter 形式追加。openTelemetry、厂商导出器、原始控制台捕获以及 trace/metric 探索同样不在本 slice 内。约束还包括：脱敏必须先于持久化执行；SQLite 崩溃可能丢失排队未刷盘事件；共享未来关系型数据库时多实例并发迁移需提供者级锁定方案。
从源码结构看（FR-023/SC-012），核心 Elsa.Diagnostics.StructuredLogs 模块不引用任何 SQLite 专属类型，SQLite 依赖被隔离在提供者包内——这保证了"SQLite 第一个、关系型可扩展"这一设计目标的成立。
相关文档导航
- 特性规范：specs/005-structured-log-persistence/spec.md
- 数据模型：specs/005-structured-log-persistence/data-model.md
- 持久化契约：specs/005-structured-log-persistence/contracts/persistence-contract.md
- 快速上手：specs/005-structured-log-persistence/quickstart.md
- 实施任务清单：specs/005-structured-log-persistence/tasks.md
- 模块 README：src/modules/Elsa.Diagnostics.StructuredLogs/README.md