diff --git a/AGENTS.md b/AGENTS.md index edefa8c..faa13bb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -81,7 +81,7 @@ - A04须截留真实task.assign并独立确认LEASED/未授权,再强杀旧节点、Wait确认退出、新节点等原租约自然到期;不能把已送达未开始替代未送达。COMMIT故障工具只在测试二进制:BEFORE阻止COMMIT,AFTER核验真实MySQL成功包后丢确认;独立连接回读,原身份/候选恢复,禁止在生产增加故障开关或将HTTP丢回执替代数据库提交不明。 - W16下一轮时钟只能在PENDING结果已保存等待后存在,三方竞争中的结果参与者是旧结果重放;新结果/未知/冲突另有在途查询回归。保持原受理时间、事件时间、截止与预算,不能凭有限入队顺序声称穷尽线程交错。 -- P6进程观测使用internal/observability固定指标目录,拒绝未知名称、类型和动态标签,不能引入任务/租户/业务键/凭据维度。ORBIT_METRICS_ADDR默认空(关闭),显式启用独立采集监听;业务端口不得暴露全局指标。渲染和网络I/O不持有全局记录锁,每个直方图局部快照保持桶/count/sum一致,最多2个并发抓取。 +- P6进程观测使用internal/observability固定指标目录,拒绝未知名称、类型和动态标签,不能引入任务/租户/业务键/凭据维度。ORBIT_METRICS_ADDR默认空(关闭),显式启用独立采集监听;业务端口不得暴露全局指标。渲染和网络I/O不持有全局记录锁,每个直方图局部快照保持桶/count/sum一致,最多2个并发抓取。启用指标后的5秒后台采样使用000026固定索引读取任务、交付和流程状态,每个状态最多10000条、900ms数据库截止;采样失败保留最后成功状态并独立记录错误/超时,不能影响业务事务或把有限样本当全局精确数。 - HTTP成功升级在Hijack完成时记录握手,WS处理器存续计入独立连接指标;保持1xx临时响应、Flush、Hijack与原panic行为。DB池取本地Stats,抓取不查询业务库;后台批次success不是业务任务成功量。观测丢样或失败不得改变权威事务及恢复决定。 -- 进程指标、格式验证、告警规则和标准回归证据见docs/reports/phase-6-observability.md。仍须补受理/授权/完成与流程推进延迟、持久吞吐/积压、重试/核查/恢复量、发送队列、实际采集告警及负载开销,不能把进程观测通过写成P6已完成。普通清理仍须保护有效幂等、未确认事件、未知结果和未终结流程。 +- 进程指标、固定状态采样、格式验证、告警规则和标准回归证据见docs/reports/phase-6-observability.md及docs/reports/phase-6-business-observability.md。仍须补受理/授权/完成与流程推进延迟、持久吞吐/积压的全局聚合、重试/核查/恢复量、发送队列、实际采集告警及负载开销,不能把进程观测通过写成P6已完成。普通清理仍须保护有效幂等、未确认事件、未知结果和未终结流程。 - 凭据轮换保留credential_id及订阅/Worker授权绑定,在同一事务替换摘要、固定到期时间、幂等回执和管理审计;新令牌必须重新建立会话,认证路径每次重新核对凭据状态。撤销必须让认证失效并保留原任务、订阅、交付、会话和审计证据,允许在应用或命名空间停用后执行;重复应用同一固定操作幂等返回。管理CLI先把固定操作文件写成0600,再提交;提交不明时保留文件并用原文件重放,不能重新生成秘密。日志、审计摘要、指标和业务表不得保存秘密原文或摘要。不要把轮换同一凭据ID解释成历史令牌仍有效,也不得级联删除订阅、任务或事件。 diff --git a/cmd/orbit/main.go b/cmd/orbit/main.go index 665fc7e..ad803d1 100644 --- a/cmd/orbit/main.go +++ b/cmd/orbit/main.go @@ -192,6 +192,25 @@ func run() error { case <-ticker.C: metrics.RecordDatabase(db.Stats()) metrics.RecordProcess() + started := time.Now() + call, cancel := context.WithTimeout(metricsCtx, 900*time.Millisecond) + snapshot, err := mysql.ReadOperationalSnapshot(call, db) + cancel() + metrics.RecordBusinessSnapshot(observability.BusinessSnapshot{ + TaskStates: snapshot.TaskStates, TaskOldestAgeSeconds: snapshot.TaskOldestAgeSeconds, + DeliveryStates: snapshot.DeliveryStates, DeliveryOldestAgeSeconds: snapshot.DeliveryOldestAgeSeconds, + WorkflowStates: snapshot.WorkflowStates, WorkflowOldestAgeSeconds: snapshot.WorkflowOldestAgeSeconds, + Truncated: snapshot.Truncated, + }, started, err) + if err != nil && metricsCtx.Err() == nil { + reason := "query_error" + if errors.Is(err, context.DeadlineExceeded) { + reason = "timeout" + } else if errors.Is(err, context.Canceled) { + reason = "cancelled" + } + slog.Error("business metrics snapshot failed", "node_id", node, "reason", reason) + } } } }() diff --git a/deploy/monitoring/orbit.rules.test.yml b/deploy/monitoring/orbit.rules.test.yml index cf97dd4..18ce0a0 100644 --- a/deploy/monitoring/orbit.rules.test.yml +++ b/deploy/monitoring/orbit.rules.test.yml @@ -22,6 +22,8 @@ tests: values: '0+0x180' - series: 'orbit_runtime_batch_last_success_timestamp_seconds{job="orbit",instance="node-a",component="workflow"}' values: '0+0x180' + - series: 'orbit_business_snapshot_last_success_timestamp_seconds{job="orbit",instance="node-a"}' + values: '0+0x180' alert_rule_test: - eval_time: 60s alertname: OrbitRuntimeBatchStalled @@ -33,6 +35,13 @@ tests: exp_annotations: summary: "Orbit background component has no recent successful batch" description: "Inspect database availability, schema readiness and batch errors before changing limits or retrying business effects." + - eval_time: 130s + alertname: OrbitBusinessSnapshotStalled + exp_alerts: + - exp_labels: {job: orbit, instance: node-a, severity: warning} + exp_annotations: + summary: "Orbit has no recent successful business-state snapshot" + description: "Inspect the bounded sampler timeout/error counter and MySQL health; the last state gauges may be stale." - name: healthy_batches_and_waits interval: 1s input_series: @@ -48,12 +57,18 @@ tests: values: '0+0x400' - series: 'orbit_observability_rejected_samples_total{job="orbit",instance="node-a"}' values: '0+0x400' + - series: 'orbit_business_snapshot_last_success_timestamp_seconds{job="orbit",instance="node-a"}' + values: '0+1x400' + - series: 'orbit_business_snapshot_truncated{job="orbit",instance="node-a"}' + values: '0+0x400' alert_rule_test: - {eval_time: 400s, alertname: OrbitMetricsUnavailable, exp_alerts: []} - {eval_time: 400s, alertname: OrbitRuntimeBatchStalled, exp_alerts: []} - {eval_time: 400s, alertname: OrbitRuntimeBatchErrors, exp_alerts: []} - {eval_time: 400s, alertname: OrbitDBPoolWaiting, exp_alerts: []} - {eval_time: 400s, alertname: OrbitInvalidMetricUpdates, exp_alerts: []} + - {eval_time: 400s, alertname: OrbitBusinessSnapshotStalled, exp_alerts: []} + - {eval_time: 400s, alertname: OrbitBusinessSnapshotTruncated, exp_alerts: []} - name: failing_batches_waits_and_invalid_samples_then_recover interval: 1s input_series: @@ -63,6 +78,10 @@ tests: values: '0+1x400 400+0x400' - series: 'orbit_observability_rejected_samples_total{job="orbit",instance="node-a"}' values: '0+1x400 400+0x400' + - series: 'orbit_business_snapshot_last_success_timestamp_seconds{job="orbit",instance="node-a"}' + values: '0+0x800' + - series: 'orbit_business_snapshot_truncated{job="orbit",instance="node-a"}' + values: '1+0x400 0+0x400' alert_rule_test: - eval_time: 400s alertname: OrbitRuntimeBatchErrors @@ -85,6 +104,14 @@ tests: exp_annotations: summary: "Orbit rejected an internal metrics update" description: "Check instrumentation changes for unsupported names, types, labels or counter regressions." + - eval_time: 400s + alertname: OrbitBusinessSnapshotTruncated + exp_alerts: + - exp_labels: {job: orbit, instance: node-a, severity: warning} + exp_annotations: + summary: "Orbit business-state snapshot reached its sample cap" + description: "State gauges are bounded samples, not exact global counts; use authoritative database queries and capacity review." - {eval_time: 800s, alertname: OrbitRuntimeBatchErrors, exp_alerts: []} - {eval_time: 800s, alertname: OrbitDBPoolWaiting, exp_alerts: []} - {eval_time: 800s, alertname: OrbitInvalidMetricUpdates, exp_alerts: []} + - {eval_time: 800s, alertname: OrbitBusinessSnapshotTruncated, exp_alerts: []} diff --git a/deploy/monitoring/orbit.rules.yml b/deploy/monitoring/orbit.rules.yml index ae1cc87..90292a3 100644 --- a/deploy/monitoring/orbit.rules.yml +++ b/deploy/monitoring/orbit.rules.yml @@ -25,6 +25,22 @@ groups: annotations: summary: "Orbit background batches are failing" description: "Use process logs and durable task/workflow audit to investigate. Batch errors are not business failures." + - alert: OrbitBusinessSnapshotStalled + expr: (time() - orbit_business_snapshot_last_success_timestamp_seconds{job="orbit"} > 30) and on (job, instance) (time() - orbit_process_start_time_seconds{job="orbit"} > 60) + for: 1m + labels: + severity: warning + annotations: + summary: "Orbit has no recent successful business-state snapshot" + description: "Inspect the bounded sampler timeout/error counter and MySQL health; the last state gauges may be stale." + - alert: OrbitBusinessSnapshotTruncated + expr: orbit_business_snapshot_truncated{job="orbit"} == 1 + for: 5m + labels: + severity: warning + annotations: + summary: "Orbit business-state snapshot reached its sample cap" + description: "State gauges are bounded samples, not exact global counts; use authoritative database queries and capacity review." - alert: OrbitDBPoolWaiting expr: rate(orbit_db_wait_duration_seconds_total{job="orbit"}[5m]) > 0.1 for: 5m diff --git a/docs/07-data-model.md b/docs/07-data-model.md index 5d43074..04034cc 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -1,6 +1,6 @@ # 数据模型与实现基线 -更新:2026-09-18。数据库统一禁止外键。当前已实现任务执行、安全重试、运行中取消、计划调度、队列限速/背压和应用配额;优先级及节点公平领取已实现;可靠订阅核心、HTTP/WSS、管理与消费 SDK 已实现,独立 Inbox 场景已实现,流程编排继续开发。 +更新:2026-09-18。数据库统一禁止外键。当前迁移版本为26;000026只增加固定业务观测读取所需索引,不增加实体或字段。当前已实现任务执行、安全重试、运行中取消、计划调度、队列限速/背压和应用配额;优先级及节点公平领取已实现;可靠订阅核心、HTTP/WSS、管理与消费 SDK 已实现,独立 Inbox 场景已实现,流程编排继续开发。 ## 1. 模块与依赖 @@ -245,3 +245,7 @@ Event.WorkflowID为只读STORED生成列,从原不可变事件派生流程归 新运行器将PENDING信号的消费、拒绝或过期与原状态事件同事务保存,批量拒绝分32条追加但整笔事务原子提交。公开查询核对固定版本全部队列、步骤/信号归属及历史激活。原始事件按数据库记录时间分页,版本与时间均不构成可靠消费水位。详见[状态时间线接口](api/workflow-events.md)与[验证报告](reports/phase-5-events.md)。 升级先停止旧写入节点,应用000025再启新版本,防止混合节点继续遗漏信号转换;生成列和索引涉及历史行扫描与可能的表重建,需预演维护窗口、磁盘和锁等待。严格SQL模式下先校验十六进制字符再转换,异常历史范围保留NULL,不清理原记录。 + +## 有界业务观测索引(000026) + +000026只增加三组读取索引:`pulse_tasks(state,available_at,id)`、`orbit_deliveries(state,available_at,id)`和`weave_instances(state,updated_at,id)`。它们服务于后台有界观测,不改变任何模型字段、状态机、事务边界或授权规则,也不包含外键或级联。采样器只使用代码固定的表、索引、状态和列名,每个状态最多读取10000条;结果达到上限时必须通过指标暴露截断,不能把样本数解释为全库精确数量。迁移后仍要求服务层 `CheckSchema` 验证零外键,真实升级见[有界业务采样报告](reports/phase-6-business-observability.md)。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index a892cb7..ba4f5e2 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -59,7 +59,7 @@ P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告]( ### P6 首批实施顺序 1. 已补[首批可复现故障工具](reports/phase-6-fault-boundaries.md):A04“派发已提交但消息尚未送达”、W01/W13事务提交结果不明,以及W16最终回执/旧PENDING结果重放/下一轮竞争。每个切点必须有实际数据库或进程证据,不能用响应错误推断事务是否已提交,也不能用普通SIGNAL竞争替代最终查询场景。 -2. 已实现[基础运行观测](api/observability.md):HTTP/WS、数据库连接池、后台批次和进程指标,以及可离线验证的告警规则;独立采集监听默认关闭,固定指标目录拒绝动态标签。仍需受理/授权/完成阶段延迟与持久吞吐、流程推进延迟、重试/核查/恢复量、待处理与到期积压、发送队列及生产采集/通知验证。 +2. 已实现[基础运行观测](api/observability.md)及[有界业务状态采样](reports/phase-6-business-observability.md):HTTP/WS、数据库连接池、后台批次、固定状态计数、最早年龄、截断/新鲜度指标和可离线验证的告警规则;独立采集监听默认关闭,固定指标目录拒绝动态标签。仍需受理/授权/完成阶段延迟与持久吞吐、流程推进延迟、重试/核查/恢复量、全局发送队列、多节点聚合及生产采集/通知验证。 3. 已补齐受保护管理环境中的固定凭据操作文件、轮换、撤销、幂等回执和脱敏管理审计;轮换保留授权绑定并要求新令牌重新建立会话,撤销允许在停用范围内执行且不改写历史记录。仍需补齐提交不明故障切点、多节点安全重绑演练、凭据最小权限配置核对和生产密钥托管/审计演练。 4. 实现保留/归档与受保护记录核对,再执行备份恢复、兼容升级和六类容量负载。未确认交付、有效幂等身份、当前/历史执行证据和未终结流程不能被普通清理破坏。生产HA及RPO/RTO继续按实际拓扑验证,不缩减为本机多进程检查。 diff --git a/docs/README.md b/docs/README.md index d8fb967..4e2b089 100644 --- a/docs/README.md +++ b/docs/README.md @@ -2,7 +2,7 @@ - 文档版本:0.2 - 整理日期:2026-09-17 -- 阶段:P0–P5平台功能已通过阶段回归;[P4/P5原范围核对](reports/phase-4-5.md)和[首批故障边界](reports/phase-6-fault-boundaries.md)保留证据边界。P6已补进程观测和固定操作文件凭据轮换/撤销;生产积压、阶段延迟、提交不明切点、归档备份、容量与发布交付继续推进,尚未完成全平台生产验收,数据库保持无外键。 +- 阶段:P0–P5平台功能已通过阶段回归;[P4/P5原范围核对](reports/phase-4-5.md)和[首批故障边界](reports/phase-6-fault-boundaries.md)保留证据边界。P6已补进程观测、有界业务状态采样和固定操作文件凭据轮换/撤销;生产积压全局聚合、阶段延迟、提交不明切点、归档备份、容量与发布交付继续推进,尚未完成全平台生产验收,数据库保持无外键。 - 适用范围:Orbit 统一平台,包含 Pulse 任务调度模块、Weave 流程编排模块与公共能力。平台面向多个业务系统复用。 ## 阅读顺序 diff --git a/docs/api/observability.md b/docs/api/observability.md index d02ca52..9879747 100644 --- a/docs/api/observability.md +++ b/docs/api/observability.md @@ -1,6 +1,6 @@ # 运行指标与告警接入 -状态:P6第一批进程观测已实现,指标、采集入口和下方规则均有回归。任务/流程的持久积压和业务阶段延迟仍待补齐;完整生产观测尚未验收。 +状态:P6第一批进程观测及有界业务状态采样已实现,指标、采集入口和下方规则均有回归。采样只提供每节点有限快照,完整生产观测尚未验收。 ## 1. 启用独立采集端口 @@ -25,7 +25,7 @@ curl -fsS http://127.0.0.1:9090/metrics 仅 `GET /metrics`(及标准HEAD)提供采集,查询参数返回400。正文使用[Prometheus文本格式0.0.4](https://prometheus.io/docs/instrumenting/exposition_formats/),Content-Type为 `text/plain; version=0.0.4; charset=utf-8`,Cache-Control为no-store。输出是本节点的运行快照,没有跨节点或跨指标原子快照保证。 -每节点同时最多2个采集请求,超出返回503及 `Retry-After: 1`,不排无限等待队列。独立HTTP服务器的读头/读/写/空闲超时分别为2/3/5/15秒,头部限制8 KiB。采集只访问进程内存,不查询业务表;DB池快照在启动时及之后每5秒读取 `database/sql.DBStats`,该调用不执行SQL。 +每节点同时最多2个采集请求,超出返回503及 `Retry-After: 1`,不排无限等待队列。独立HTTP服务器的读头/读/写/空闲超时分别为2/3/5/15秒,头部限制8 KiB。每次HTTP采集只访问进程内存,不查询业务表;DB池快照在启动时及之后每5秒读取 `database/sql.DBStats`,该调用不执行SQL。指标启用后同一个5秒后台采样器另外执行一次有界业务快照:查询固定状态、固定表和固定索引,单状态最多读取10000条,数据库调用使用900ms截止时间;失败只增加错误/超时计数并保留上次成功值,不改变业务事务。 所有指标名称、类型、标签名和取值在代码目录中固定。未知名称、错误类型、非法标签、非有限数值和计数器倒退会被丢弃,并增加 `orbit_observability_rejected_samples_total`。新增指标需要修改目录和回归,不允许业务方动态注册标签。HTTP标签为8种方法、12种路由分类和6种状态分类,两组HTTP族各最多576种标签组合。禁止将任务/流程/租户ID、业务键、URL、参数、原始错误或凭据写入指标。 @@ -47,6 +47,16 @@ curl -fsS http://127.0.0.1:9090/metrics | `orbit_db_connections_open` / `orbit_db_connections_in_use` / `orbit_db_connections_idle` | gauge | 本节点MySQL连接池当前连接数 | | `orbit_db_connections_max_open` | gauge | 池配置的最大连接数 | | `orbit_db_wait_count_total` / `orbit_db_wait_duration_seconds_total` | counter | 连接池累计等待次数/秒数,不是数据库锁等待 | +| `orbit_pulse_tasks_state{state}` | gauge | Pulse任务固定状态的有界计数;单状态达到10000时只保留上限并置截断标记 | +| `orbit_pulse_tasks_oldest_age_seconds` | gauge | WAITING/LEASED/RUNNING/NEEDS_REVIEW样本中最早可处理时间的年龄 | +| `orbit_result_deliveries_state{state}` | gauge | 结果交付固定状态的有界计数 | +| `orbit_result_delivery_oldest_age_seconds` | gauge | PENDING/LEASED交付样本中最早可处理时间的年龄 | +| `orbit_weave_workflows_state{state}` | gauge | Weave实例固定状态的有界计数 | +| `orbit_weave_workflow_oldest_age_seconds` | gauge | RUNNING/WAITING/CANCELLING/NEEDS_REVIEW实例样本中最早更新时间的年龄 | +| `orbit_business_snapshot_total{outcome}` | counter | 有界业务快照成功、错误或超时次数 | +| `orbit_business_snapshot_duration_seconds` | histogram | 有界业务快照耗时;包含数据库读取 | +| `orbit_business_snapshot_last_success_timestamp_seconds` | gauge | 最近一次完整成功快照的Unix时间;失败不会推进 | +| `orbit_business_snapshot_truncated` | gauge | 任一固定状态达到10000上限时为1,否则为0 | | `orbit_process_start_time_seconds` / `orbit_process_uptime_seconds` | gauge | 本次指标注册器初始化Unix时间/单调时钟经过的秒数;初始化发生在数据库启动校验之后 | | `orbit_go_goroutines` | gauge | Go协程数,按5秒采样 | | `orbit_build_info` | gauge | 恒为1,go_version仅取当前编译运行时版本;不是Orbit发布版本 | @@ -73,11 +83,13 @@ scrape_configs: 示例地址仅适用于同机采集;远程采集需替换为受保护的内部目标。平台运行不依赖采集器、Prometheus或额外数据库。 -[五条告警规则](../../deploy/monitoring/orbit.rules.yml)和[离线规则测试](../../deploy/monitoring/orbit.rules.test.yml)包含以下场景: +[七条告警规则](../../deploy/monitoring/orbit.rules.yml)和[离线规则测试](../../deploy/monitoring/orbit.rules.test.yml)包含以下场景: - 目标连续2分钟采集失败:检查指标监听、网络和进程,不直接判断业务任务停止。 - 启动60秒后,某后台组件超过30秒没有成功批次并持续1分钟:结合readiness、批次错误和数据库状态调查。 - 批次错误速率持续为正2分钟:查日志和原任务/流程审计,不把数据库故障改判成业务失败。 +- 有界业务快照连续没有成功超过30秒(启动宽限60秒):检查采样超时/错误和MySQL健康;已有状态值可能已经陈旧。 +- 固定状态采样连续达到10000条上限5分钟:状态指标只是有限样本,转用权威数据库查询并进行容量核查。 - 池累计等待时间增长超过0.1秒/秒并持续5分钟:联查锁等待、慢SQL、CPU和存储,不自动提高连接数。 - 内部非法指标更新持续出现:检查观测代码契约,不能忽略丢弃的样本。 @@ -88,4 +100,4 @@ promtool check rules deploy/monitoring/orbit.rules.yml promtool test rules deploy/monitoring/orbit.rules.test.yml ``` -HTTP失败率和延迟可按route/status_class统计;受理接口响应成功与已完成业务应分别判断。可靠结果量、排队/授权/完成延迟、流程Inbox积压、发送队列、磁盘/复制等仍须补充专门指标;具体当前证据见[验证报告](../reports/phase-6-observability.md)。 +HTTP失败率和延迟可按route/status_class统计;受理接口响应成功与已完成业务应分别判断。业务状态指标是本节点有限样本,不能相加作为全局精确数量;应结合 `orbit_business_snapshot_last_success_timestamp_seconds` 和 `orbit_business_snapshot_truncated` 判断新鲜度与完整性。可靠结果量、排队/授权/完成阶段延迟、流程Inbox推进延迟、发送队列、磁盘/复制等仍须补充专门指标;具体当前证据见[进程观测报告](../reports/phase-6-observability.md)和[有界业务采样报告](../reports/phase-6-business-observability.md)。 diff --git a/docs/reports/phase-6-business-observability.md b/docs/reports/phase-6-business-observability.md new file mode 100644 index 0000000..3d04854 --- /dev/null +++ b/docs/reports/phase-6-business-observability.md @@ -0,0 +1,24 @@ +# P6:有界业务状态采样 + +日期:2026-09-18。本文记录第二批运行观测实现的边界和回归证据。它补充进程级批次指标,用于回答“本节点看到哪些固定状态、最早样本等待多久、采样是否完整”,不把有限样本当作全局业务统计,也不宣称生产容量或高可用已经验收。 + +## 已实现 + +- 迁移 `000026_operational_metrics` 为 `pulse_tasks`、`orbit_deliveries`、`weave_instances` 增加固定前缀索引;不增加模型字段,不引入外键或级联。 +- [存储采样器](../../internal/storage/mysql/operational_metrics.go)只拼接代码内固定的表、索引、列名和状态集合。每个状态使用索引读取最多10001条,再将结果封顶到10000;至少一类封顶时 `orbit_business_snapshot_truncated=1`。 +- 采样器先读取数据库UTC时间,再读取任务、结果交付和流程实例固定状态计数及最早未完成年龄。所有调用由独立900ms上下文限制;取消、数据库错误和超时都返回失败,不修改持久化数据。 +- [业务指标记录器](../../internal/observability/business.go)将成功、错误、超时分别计数并记录耗时。失败样本不覆盖最后一次成功状态,同时保留最后成功时间,方便告警区分“旧值可见”和“本轮采集失败”。标签基数和状态取值均在目录中固定。 +- [进程入口](../../cmd/orbit/main.go)仅在显式启用指标时每5秒后台采样;`GET /metrics`仍只渲染进程内存,不在抓取请求中执行SQL。采样错误进入脱敏结构化日志,不能改变任务、交付或流程事务。 + +## 回归证据 + +- [业务记录器单测](../../internal/observability/metrics_test.go)验证成功状态、年龄、截断、成功计数和失败后保留最后成功值;超时计数单独增加。 +- [真实MySQL专项](../../internal/storage/mysql/operational_metrics_integration_test.go)验证迁移索引真实存在、数据库外键为0、任务状态被采样、固定状态集合存在、空交付/流程状态不误报,并验证已取消上下文被拒绝。 +- `go test ./...`覆盖编译和普通单测;真实MySQL专项由标准`GO=... DOCKER=... sh scripts/test-mysql.sh`执行,未设置`ORBIT_TEST_MYSQL_DSN`时专项会跳过,不能把跳过当通过。最终标准脚本在Go 1.27.1、darwin/arm64、Docker MySQL 8.4.10环境退出码为0;其中存储/迁移/事务组256.539秒,指标开关进程组5.204秒,流程运行器全停恢复49.333秒,SDK与两个独立业务示例均通过。专项日志确认`TestIntegrationOperationalSnapshot`、`TestIntegrationNoForeignKeysUpgrade`、`TestIntegrationCredentialOperations`及`TestIntegrationCredentialCommitFaults`均PASS。 +- 当前固定目录导出的指标夹具为929159字节,`promtool check metrics`成功;`promtool check rules deploy/monitoring/orbit.rules.yml`报告7条规则成功,`promtool test rules deploy/monitoring/orbit.rules.test.yml`成功。标准验证镜像仅用于离线检查,平台运行不依赖Prometheus。 + +## 解释和边界 + +每个Orbit节点都读取自己的有限样本;多节点指标不能直接相加获得全局精确积压。达到10000上限、采样超时、节点重启或节点间时间不同都会影响解释,必须结合截断标记、最后成功时间、数据库权威查询和审计判断。最早年龄使用任务/交付的`available_at`及流程的`updated_at`,它是告警线索,不是端到端受理、授权或完成延迟。 + +当前仍未完成持久化吞吐与阶段延迟、流程Inbox推进延迟、重试/核查/恢复量、发送队列全局聚合、生产采集通知演练、锁/磁盘/复制指标和多故障域容量验证。此批实现不能替代保留、备份恢复、HA、RPO/RTO及发布验收。 diff --git a/docs/reports/phase-6-observability.md b/docs/reports/phase-6-observability.md index eccca02..c908ca0 100644 --- a/docs/reports/phase-6-observability.md +++ b/docs/reports/phase-6-observability.md @@ -1,6 +1,6 @@ # P6:进程指标、采集隔离与告警规则 -日期:2026-09-18。范围为运行观测第一批,交付HTTP/WS、连接池、后台批次、进程指标和可离线验证的告警规则。P6仍在进行,任务/流程持久积压和阶段延迟没有被本批替代。 +日期:2026-09-18。范围为运行观测第一批,交付HTTP/WS、连接池、后台批次、进程指标和可离线验证的告警规则。本文记录第一批完成时的迁移版本25;后续000026有界业务状态采样见[专项报告](phase-6-business-observability.md)。P6仍在进行,完整生产观测没有被任一批次替代。 ## 实现和边界 @@ -24,7 +24,7 @@ ## 独立格式与规则验证 -全量有限HTTP组合的文本夹具为926666字节,使用本地已有官方promtool 3.5.0(linux/arm64,revision 8be3a9560fbdd18a94dedec4b747c35178177202)执行`check metrics`,退出0且无格式/命名告警。固定验证镜像摘要为`prom/prometheus@sha256:63805ebb8d2b3920190daf1cb14a60871b16fd38bed42b857a3182bc621f4996`,运行时关闭网络;此版本仅描述实际验证器,不是生产部署版本建议。 +第一批全量有限HTTP组合的文本夹具为926666字节,使用本地已有官方promtool 3.5.0(linux/arm64,revision 8be3a9560fbdd18a94dedec4b747c35178177202)执行`check metrics`,退出0且无格式/命名告警。固定验证镜像摘要为`prom/prometheus@sha256:63805ebb8d2b3920190daf1cb14a60871b16fd38bed42b857a3182bc621f4996`,运行时关闭网络;此版本仅描述实际验证器,不是生产部署版本建议。当前新增业务状态目录的夹具和校验见专项报告。 五条告警规则通过`check rules`;[规则测试](../../deploy/monitoring/orbit.rules.test.yml)通过`test rules`,验证目标失联、启动宽限、组件停滞、健康不报警、错误/等待/非法采样触发及恢复后的解除。 diff --git a/internal/observability/business.go b/internal/observability/business.go new file mode 100644 index 0000000..b8bd9fa --- /dev/null +++ b/internal/observability/business.go @@ -0,0 +1,65 @@ +package observability + +import ( + "context" + "errors" + "time" +) + +// BusinessSnapshot 是存储层返回的固定基数业务快照。 +// Map 的键只由存储层固定状态集合产生,Registry 仍只向目录中的标签写值。 +type BusinessSnapshot struct { + // TaskStates Pulse 任务状态的有界计数。 + TaskStates map[string]uint64 + // TaskOldestAgeSeconds 未完成任务中最早可处理记录的年龄。 + TaskOldestAgeSeconds float64 + // DeliveryStates 结果交付状态的有界计数。 + DeliveryStates map[string]uint64 + // DeliveryOldestAgeSeconds 待处理交付中最早可处理记录的年龄。 + DeliveryOldestAgeSeconds float64 + // WorkflowStates Weave 实例状态的有界计数。 + WorkflowStates map[string]uint64 + // WorkflowOldestAgeSeconds 未终结流程中最早更新时间的年龄。 + WorkflowOldestAgeSeconds float64 + // Truncated 是否至少有一类状态达到采样上限。 + Truncated bool +} + +// RecordBusinessSnapshot records one bounded database sample. A failed sample +// does not overwrite the last successful state, so stale data remains visible +// together with an error counter and alertable last-success timestamp. +func (r *Registry) RecordBusinessSnapshot(snapshot BusinessSnapshot, started time.Time, err error) { + if r == nil { + return + } + outcome := "success" + if err != nil { + outcome = "error" + if errors.Is(err, context.DeadlineExceeded) { + outcome = "timeout" + } + } + r.AddCounter("orbit_business_snapshot_total", 1, Labels{"outcome": outcome}) + r.ObserveHistogram("orbit_business_snapshot_duration_seconds", time.Since(started).Seconds(), nil) + if err != nil { + return + } + for _, state := range taskStates { + r.SetGauge("orbit_pulse_tasks_state", float64(snapshot.TaskStates[state]), Labels{"state": state}) + } + for _, state := range deliveryStates { + r.SetGauge("orbit_result_deliveries_state", float64(snapshot.DeliveryStates[state]), Labels{"state": state}) + } + for _, state := range workflowStates { + r.SetGauge("orbit_weave_workflows_state", float64(snapshot.WorkflowStates[state]), Labels{"state": state}) + } + r.SetGauge("orbit_pulse_tasks_oldest_age_seconds", snapshot.TaskOldestAgeSeconds, nil) + r.SetGauge("orbit_result_delivery_oldest_age_seconds", snapshot.DeliveryOldestAgeSeconds, nil) + r.SetGauge("orbit_weave_workflow_oldest_age_seconds", snapshot.WorkflowOldestAgeSeconds, nil) + truncated := 0.0 + if snapshot.Truncated { + truncated = 1 + } + r.SetGauge("orbit_business_snapshot_truncated", truncated, nil) + r.SetGauge("orbit_business_snapshot_last_success_timestamp_seconds", float64(time.Now().Unix()), nil) +} diff --git a/internal/observability/catalog.go b/internal/observability/catalog.go index 8179308..6c0d1bf 100644 --- a/internal/observability/catalog.go +++ b/internal/observability/catalog.go @@ -6,6 +6,9 @@ var methods = []string{"GET", "POST", "PATCH", "PUT", "DELETE", "HEAD", "OPTIONS var routes = []string{"health_live", "health_ready", "worker_connect", "event_connect", "tasks", "queues", "workflows", "workflow_definitions", "schedules", "subscriptions", "application_resources", "other"} var statusClasses = []string{"1xx", "2xx", "3xx", "4xx", "5xx", "other"} var components = []string{"workflow", "schedule", "execution_recovery"} +var taskStates = []string{"WAITING", "LEASED", "RUNNING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED"} +var deliveryStates = []string{"PENDING", "LEASED", "ACKED", "EXPIRED"} +var workflowStates = []string{"RUNNING", "WAITING", "CANCELLING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED"} // Every exported family is declared here, including each allowed label value. // Adding a metric therefore requires a code review; no user identifier is valid. @@ -33,5 +36,15 @@ func catalog() map[string]descriptor { } c["orbit_db_wait_count_total"] = descriptor{counter, "Cumulative waits for a MySQL connection", nil} c["orbit_db_wait_duration_seconds_total"] = descriptor{counter, "Cumulative seconds waiting for a MySQL connection", nil} + c["orbit_pulse_tasks_state"] = descriptor{gauge, "Bounded count of Pulse tasks by durable state", map[string][]string{"state": taskStates}} + c["orbit_pulse_tasks_oldest_age_seconds"] = descriptor{gauge, "Age of the oldest sampled unfinished Pulse task", nil} + c["orbit_result_deliveries_state"] = descriptor{gauge, "Bounded count of result deliveries by durable state", map[string][]string{"state": deliveryStates}} + c["orbit_result_delivery_oldest_age_seconds"] = descriptor{gauge, "Age of the oldest sampled pending result delivery", nil} + c["orbit_weave_workflows_state"] = descriptor{gauge, "Bounded count of Weave instances by durable state", map[string][]string{"state": workflowStates}} + c["orbit_weave_workflow_oldest_age_seconds"] = descriptor{gauge, "Age of the oldest sampled unfinished Weave instance", nil} + c["orbit_business_snapshot_total"] = descriptor{counter, "Completed bounded business-state snapshots by outcome", map[string][]string{"outcome": {"success", "error", "timeout"}}} + c["orbit_business_snapshot_duration_seconds"] = descriptor{histogram, "Bounded business-state snapshot duration in seconds", nil} + c["orbit_business_snapshot_last_success_timestamp_seconds"] = descriptor{gauge, "Unix time of the last successful bounded business-state snapshot", nil} + c["orbit_business_snapshot_truncated"] = descriptor{gauge, "Whether a bounded business-state sample hit its per-state cap", nil} return c } diff --git a/internal/observability/metrics_test.go b/internal/observability/metrics_test.go index ab78825..0fb2a76 100644 --- a/internal/observability/metrics_test.go +++ b/internal/observability/metrics_test.go @@ -112,6 +112,42 @@ func TestRuntimeObservationDoesNotExposeErrors(t *testing.T) { disabled.RecordBatch("workflow", time.Now(), nil) disabled.RecordDatabase(sql.DBStats{}) } + +func TestBusinessSnapshotKeepsLastGoodStateOnFailure(t *testing.T) { + r := NewRegistry() + started := time.Now().Add(-25 * time.Millisecond) + r.RecordBusinessSnapshot(BusinessSnapshot{ + TaskStates: map[string]uint64{"WAITING": 3}, + TaskOldestAgeSeconds: 12.5, + DeliveryStates: map[string]uint64{"PENDING": 2}, + DeliveryOldestAgeSeconds: 8.25, + WorkflowStates: map[string]uint64{"RUNNING": 1}, + WorkflowOldestAgeSeconds: 4.75, + Truncated: true, + }, started, nil) + first := r.Render() + for _, want := range []string{ + `orbit_pulse_tasks_state{state="WAITING"} 3`, + `orbit_result_deliveries_state{state="PENDING"} 2`, + `orbit_weave_workflows_state{state="RUNNING"} 1`, + "orbit_pulse_tasks_oldest_age_seconds 12.5", + "orbit_business_snapshot_truncated 1", + `orbit_business_snapshot_total{outcome="success"} 1`, + } { + if !strings.Contains(first, want) { + t.Fatalf("missing business sample %q", want) + } + } + lastSuccess := r.families["orbit_business_snapshot_last_success_timestamp_seconds"].series[""].value.Load() + r.RecordBusinessSnapshot(BusinessSnapshot{TaskStates: map[string]uint64{"WAITING": 99}}, time.Now(), context.DeadlineExceeded) + second := r.Render() + if !strings.Contains(second, `orbit_business_snapshot_total{outcome="timeout"} 1`) || !strings.Contains(second, `orbit_pulse_tasks_state{state="WAITING"} 3`) { + t.Fatal("failed sample overwrote good state") + } + if got := r.families["orbit_business_snapshot_last_success_timestamp_seconds"].series[""].value.Load(); got != lastSuccess { + t.Fatal("failed sample changed last success timestamp") + } +} func TestHTTPBoundedLabelsAndPanicPreserved(t *testing.T) { r := NewRegistry() h := r.Middleware(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { w.WriteHeader(202) })) diff --git a/internal/storage/mysql/application_resources_upgrade_integration_test.go b/internal/storage/mysql/application_resources_upgrade_integration_test.go index 3b46388..9fccc6f 100644 --- a/internal/storage/mysql/application_resources_upgrade_integration_test.go +++ b/internal/storage/mysql/application_resources_upgrade_integration_test.go @@ -65,6 +65,9 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) { assertApplicationCounts(t, f, 4, 2, 1) // 还原后续新增表、字段和索引,得到真实 v9 结构;现有任务内容保持不变。 f.exec(`ALTER TABLE orbit_events DROP INDEX ix_event_workflow_history,DROP COLUMN workflow_id`) + f.exec(`ALTER TABLE pulse_tasks DROP INDEX ix_task_observe_state_available`) + f.exec(`ALTER TABLE orbit_deliveries DROP INDEX ix_delivery_observe_state_available`) + f.exec(`ALTER TABLE weave_instances DROP INDEX ix_workflow_observe_state_updated`) f.exec(`DROP TABLE weave_version_scopes`) f.exec(`ALTER TABLE weave_instances DROP INDEX ix_workflow_created,DROP INDEX ix_workflow_state_created,DROP INDEX ix_workflow_business_state_created,DROP INDEX ix_workflow_definition_created,DROP INDEX ix_workflow_definition_state_created,DROP INDEX ix_workflow_definition_business_created,DROP INDEX ix_workflow_definition_business_state_created`) f.exec(`ALTER TABLE weave_signals DROP INDEX ix_signal_history`) diff --git a/internal/storage/mysql/migrations/000026_operational_metrics.up.sql b/internal/storage/mysql/migrations/000026_operational_metrics.up.sql new file mode 100644 index 0000000..2dca934 --- /dev/null +++ b/internal/storage/mysql/migrations/000026_operational_metrics.up.sql @@ -0,0 +1,10 @@ +-- 为有界运行观测提供按状态和时间的前缀索引。 +-- 采集只读取固定状态的有限样本,不依赖外键或级联关系。 +ALTER TABLE pulse_tasks + ADD KEY ix_task_observe_state_available (state, available_at, id); + +ALTER TABLE orbit_deliveries + ADD KEY ix_delivery_observe_state_available (state, available_at, id); + +ALTER TABLE weave_instances + ADD KEY ix_workflow_observe_state_updated (state, updated_at, id); diff --git a/internal/storage/mysql/no_foreign_keys_integration_test.go b/internal/storage/mysql/no_foreign_keys_integration_test.go index af84793..c2fcd96 100644 --- a/internal/storage/mysql/no_foreign_keys_integration_test.go +++ b/internal/storage/mysql/no_foreign_keys_integration_test.go @@ -89,7 +89,7 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) { indexes := func() []string { t.Helper() // 排除后续新增/替换索引;原有项在此逐项比较,v24/v25新增项由对应Query/Timeline升级测试核对。 - rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events','weave_version_scopes') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history','uq_task_retry_parent','fk_pulse_tasks_3','ix_task_queue_business_created','ix_task_queue_business_state_created','uq_weave_task_result','ix_signal_expiry','ix_signal_activation','uq_wait_winner_signal','ix_task_workflow_owner','ix_signal_history','ix_event_workflow_history','ix_workflow_created','ix_workflow_state_created','ix_workflow_business_state_created','ix_workflow_definition_created','ix_workflow_definition_state_created','ix_workflow_definition_business_created','ix_workflow_definition_business_state_created') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`) + rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events','weave_version_scopes') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history','uq_task_retry_parent','fk_pulse_tasks_3','ix_task_queue_business_created','ix_task_queue_business_state_created','uq_weave_task_result','ix_signal_expiry','ix_signal_activation','uq_wait_winner_signal','ix_task_workflow_owner','ix_signal_history','ix_event_workflow_history','ix_workflow_created','ix_workflow_state_created','ix_workflow_business_state_created','ix_workflow_definition_created','ix_workflow_definition_state_created','ix_workflow_definition_business_created','ix_workflow_definition_business_state_created','ix_task_observe_state_available','ix_delivery_observe_state_available','ix_workflow_observe_state_updated') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`) if err != nil { t.Fatal(err) } diff --git a/internal/storage/mysql/operational_metrics.go b/internal/storage/mysql/operational_metrics.go new file mode 100644 index 0000000..4873ad8 --- /dev/null +++ b/internal/storage/mysql/operational_metrics.go @@ -0,0 +1,126 @@ +package mysql + +import ( + "context" + "database/sql" + "errors" + "time" +) + +const operationalMetricSampleLimit = 10000 + +// OperationalSnapshot 是进程观测使用的有限数据库快照。 +// Counts 只包含固定状态的有界计数;达到上限时 Truncated 为 true,不能解释为精确总量。 +type OperationalSnapshot struct { + // TaskStates 是 Pulse 任务按状态的有界计数。 + TaskStates map[string]uint64 + // TaskOldestAgeSeconds 是运行中任务中最早 available_at 的年龄。 + TaskOldestAgeSeconds float64 + // DeliveryStates 是结果交付按状态的有界计数。 + DeliveryStates map[string]uint64 + // DeliveryOldestAgeSeconds 是待处理交付中最早 available_at 的年龄。 + DeliveryOldestAgeSeconds float64 + // WorkflowStates 是 Weave 实例按状态的有界计数。 + WorkflowStates map[string]uint64 + // WorkflowOldestAgeSeconds 是未终结流程中最早 updated_at 的年龄。 + WorkflowOldestAgeSeconds float64 + // Truncated 表示至少一个状态达到有界采样上限。 + Truncated bool +} + +var ( + operationalTaskStates = []string{"WAITING", "LEASED", "RUNNING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED"} + operationalDeliveryStates = []string{"PENDING", "LEASED", "ACKED", "EXPIRED"} + operationalWorkflowStates = []string{"RUNNING", "WAITING", "CANCELLING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED"} +) + +// ReadOperationalSnapshot performs only bounded, indexed reads. It is intended +// for a background sampler with a short context deadline, never for a request. +func ReadOperationalSnapshot(ctx context.Context, db *sql.DB) (OperationalSnapshot, error) { + if db == nil { + return OperationalSnapshot{}, errors.New("database is required") + } + now, err := readDatabaseNow(ctx, db) + if err != nil { + return OperationalSnapshot{}, err + } + tasks, taskTruncated, err := boundedStateCounts(ctx, db, "pulse_tasks", "ix_task_observe_state_available", operationalTaskStates) + if err != nil { + return OperationalSnapshot{}, err + } + deliveries, deliveryTruncated, err := boundedStateCounts(ctx, db, "orbit_deliveries", "ix_delivery_observe_state_available", operationalDeliveryStates) + if err != nil { + return OperationalSnapshot{}, err + } + workflows, workflowTruncated, err := boundedStateCounts(ctx, db, "weave_instances", "ix_workflow_observe_state_updated", operationalWorkflowStates) + if err != nil { + return OperationalSnapshot{}, err + } + taskAge, err := oldestAge(ctx, db, "pulse_tasks", "ix_task_observe_state_available", []string{"WAITING", "LEASED", "RUNNING", "NEEDS_REVIEW"}, "available_at", now) + if err != nil { + return OperationalSnapshot{}, err + } + deliveryAge, err := oldestAge(ctx, db, "orbit_deliveries", "ix_delivery_observe_state_available", []string{"PENDING", "LEASED"}, "available_at", now) + if err != nil { + return OperationalSnapshot{}, err + } + workflowAge, err := oldestAge(ctx, db, "weave_instances", "ix_workflow_observe_state_updated", []string{"RUNNING", "WAITING", "CANCELLING", "NEEDS_REVIEW"}, "updated_at", now) + if err != nil { + return OperationalSnapshot{}, err + } + return OperationalSnapshot{ + TaskStates: tasks, TaskOldestAgeSeconds: taskAge, + DeliveryStates: deliveries, DeliveryOldestAgeSeconds: deliveryAge, + WorkflowStates: workflows, WorkflowOldestAgeSeconds: workflowAge, + Truncated: taskTruncated || deliveryTruncated || workflowTruncated, + }, nil +} + +func readDatabaseNow(ctx context.Context, db *sql.DB) (time.Time, error) { + var now time.Time + if err := db.QueryRowContext(ctx, "SELECT UTC_TIMESTAMP(6)").Scan(&now); err != nil { + return time.Time{}, err + } + return now.UTC(), nil +} + +func boundedStateCounts(ctx context.Context, db *sql.DB, table, index string, states []string) (map[string]uint64, bool, error) { + counts := make(map[string]uint64, len(states)) + truncated := false + for _, state := range states { + query := "SELECT COUNT(*) FROM (SELECT id FROM " + table + " FORCE INDEX (" + index + ") WHERE state=? LIMIT ?) AS bounded" + var n uint64 + if err := db.QueryRowContext(ctx, query, state, operationalMetricSampleLimit+1).Scan(&n); err != nil { + return nil, false, err + } + if n > operationalMetricSampleLimit { + n = operationalMetricSampleLimit + truncated = true + } + counts[state] = n + } + return counts, truncated, nil +} + +func oldestAge(ctx context.Context, db *sql.DB, table, index string, states []string, column string, now time.Time) (float64, error) { + placeholders := "?" + args := make([]any, 0, len(states)) + args = append(args, states[0]) + for _, state := range states[1:] { + placeholders += ",?" + args = append(args, state) + } + query := "SELECT " + column + " FROM " + table + " FORCE INDEX (" + index + ") WHERE state IN (" + placeholders + ") ORDER BY " + column + ",id LIMIT 1" + var occurred time.Time + if err := db.QueryRowContext(ctx, query, args...).Scan(&occurred); err != nil { + if err == sql.ErrNoRows { + return 0, nil + } + return 0, err + } + age := now.Sub(occurred.UTC()).Seconds() + if age < 0 { + return 0, nil + } + return age, nil +} diff --git a/internal/storage/mysql/operational_metrics_integration_test.go b/internal/storage/mysql/operational_metrics_integration_test.go new file mode 100644 index 0000000..ba3a4c2 --- /dev/null +++ b/internal/storage/mysql/operational_metrics_integration_test.go @@ -0,0 +1,83 @@ +package mysql + +import ( + "context" + "os" + "testing" + "time" +) + +// TestIntegrationOperationalSnapshot verifies the bounded sampler against the +// real MySQL grammar and the indexes installed by migration 000026. +func TestIntegrationOperationalSnapshot(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("set ORBIT_TEST_MYSQL_DSN to a disposable database, or run make integration") + } + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + db, err := Open(ctx, dsn, 16, 4) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + if err := CheckSchema(ctx, db); err != nil { + t.Fatal(err) + } + for _, want := range []struct { + table string + index string + }{ + {"pulse_tasks", "ix_task_observe_state_available"}, + {"orbit_deliveries", "ix_delivery_observe_state_available"}, + {"weave_instances", "ix_workflow_observe_state_updated"}, + } { + var n int + if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name=? AND index_name=?`, want.table, want.index).Scan(&n); err != nil { + t.Fatal(err) + } + if n == 0 { + t.Fatalf("missing operational index %s.%s", want.table, want.index) + } + } + var foreignKeys int + if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.REFERENTIAL_CONSTRAINTS WHERE CONSTRAINT_SCHEMA=DATABASE()`).Scan(&foreignKeys); err != nil { + t.Fatal(err) + } + if foreignKeys != 0 { + t.Fatalf("operational migration introduced foreign keys: %d", foreignKeys) + } + + f := newExecutionFixture(t, ctx, db) + f.submit(2) + snapshot, err := ReadOperationalSnapshot(ctx, db) + if err != nil { + t.Fatal(err) + } + if got := snapshot.TaskStates["WAITING"]; got < 1 { + t.Fatalf("waiting task was not observed: %#v", snapshot.TaskStates) + } + if snapshot.TaskOldestAgeSeconds < 0 || snapshot.DeliveryOldestAgeSeconds < 0 || snapshot.WorkflowOldestAgeSeconds < 0 { + t.Fatalf("negative age in snapshot: %#v", snapshot) + } + for _, state := range []string{"WAITING", "LEASED", "RUNNING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED"} { + if _, ok := snapshot.TaskStates[state]; !ok { + t.Fatalf("missing fixed task state %q", state) + } + } + if _, ok := snapshot.DeliveryStates["PENDING"]; !ok { + t.Fatal("missing pending delivery state") + } + if _, ok := snapshot.WorkflowStates["RUNNING"]; !ok { + t.Fatal("missing running workflow state") + } + + cancelled, cancelNow := context.WithCancel(ctx) + cancelNow() + if _, err := ReadOperationalSnapshot(cancelled, db); err == nil { + t.Fatal("cancelled context unexpectedly completed snapshot") + } +}