109 lines
5.7 KiB
Go
109 lines
5.7 KiB
Go
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
|
||
// InboxStates Weave 内部事件状态的有界计数。
|
||
InboxStates map[string]uint64
|
||
// InboxOldestAgeSeconds 待消费内部事件中最早可处理记录的年龄。
|
||
InboxOldestAgeSeconds float64
|
||
// Truncated 是否至少有一类状态达到采样上限。
|
||
Truncated bool
|
||
// RecentEvents 固定事件类别在最近窗口内的有界数量。
|
||
RecentEvents map[string]uint64
|
||
// TaskTerminalLatencySeconds 最近窗口终态任务创建到完成的平均耗时样本。
|
||
TaskTerminalLatencySeconds float64
|
||
// TaskTerminalLatencyMaxSeconds 最近窗口终态任务样本最大耗时。
|
||
TaskTerminalLatencyMaxSeconds float64
|
||
// WorkflowTerminalLatencySeconds 最近窗口终态流程创建到完成的平均耗时样本。
|
||
WorkflowTerminalLatencySeconds float64
|
||
// WorkflowTerminalLatencyMaxSeconds 最近窗口终态流程样本最大耗时。
|
||
WorkflowTerminalLatencyMaxSeconds float64
|
||
// TaskAcceptToStartLatencySeconds 最近窗口任务受理到开始的平均耗时样本。
|
||
TaskAcceptToStartLatencySeconds float64
|
||
// TaskAcceptToStartLatencyMaxSeconds 最近窗口任务受理到开始的最大耗时样本。
|
||
TaskAcceptToStartLatencyMaxSeconds float64
|
||
// TaskAcceptToTerminalLatencySeconds 最近窗口任务受理到确定终态的平均耗时样本。
|
||
TaskAcceptToTerminalLatencySeconds float64
|
||
// TaskAcceptToTerminalLatencyMaxSeconds 最近窗口任务受理到确定终态的最大耗时样本。
|
||
TaskAcceptToTerminalLatencyMaxSeconds float64
|
||
// WorkflowStartToTerminalLatencySeconds 最近窗口流程开始到停止的平均耗时样本。
|
||
WorkflowStartToTerminalLatencySeconds float64
|
||
// WorkflowStartToTerminalLatencyMaxSeconds 最近窗口流程开始到停止的最大耗时样本。
|
||
WorkflowStartToTerminalLatencyMaxSeconds float64
|
||
}
|
||
|
||
// 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})
|
||
}
|
||
for _, state := range inboxStates {
|
||
r.SetGauge("orbit_weave_inbox_state", float64(snapshot.InboxStates[state]), Labels{"state": state})
|
||
}
|
||
for _, eventType := range eventTypes {
|
||
r.SetGauge("orbit_business_events_recent", float64(snapshot.RecentEvents[eventType]), Labels{"event": eventType})
|
||
}
|
||
r.SetGauge("orbit_pulse_task_terminal_latency_seconds", snapshot.TaskTerminalLatencySeconds, nil)
|
||
r.SetGauge("orbit_pulse_task_terminal_latency_max_seconds", snapshot.TaskTerminalLatencyMaxSeconds, nil)
|
||
r.SetGauge("orbit_weave_workflow_terminal_latency_seconds", snapshot.WorkflowTerminalLatencySeconds, nil)
|
||
r.SetGauge("orbit_weave_workflow_terminal_latency_max_seconds", snapshot.WorkflowTerminalLatencyMaxSeconds, nil)
|
||
r.SetGauge("orbit_pulse_task_accept_to_start_latency_seconds", snapshot.TaskAcceptToStartLatencySeconds, nil)
|
||
r.SetGauge("orbit_pulse_task_accept_to_start_latency_max_seconds", snapshot.TaskAcceptToStartLatencyMaxSeconds, nil)
|
||
r.SetGauge("orbit_pulse_task_accept_to_terminal_latency_seconds", snapshot.TaskAcceptToTerminalLatencySeconds, nil)
|
||
r.SetGauge("orbit_pulse_task_accept_to_terminal_latency_max_seconds", snapshot.TaskAcceptToTerminalLatencyMaxSeconds, nil)
|
||
r.SetGauge("orbit_weave_workflow_start_to_terminal_latency_seconds", snapshot.WorkflowStartToTerminalLatencySeconds, nil)
|
||
r.SetGauge("orbit_weave_workflow_start_to_terminal_latency_max_seconds", snapshot.WorkflowStartToTerminalLatencyMaxSeconds, nil)
|
||
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)
|
||
r.SetGauge("orbit_weave_inbox_oldest_age_seconds", snapshot.InboxOldestAgeSeconds, 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)
|
||
}
|