376 lines
15 KiB
Go
376 lines
15 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"sort"
|
||
"time"
|
||
)
|
||
|
||
const operationalMetricSampleLimit = 10000
|
||
const operationalMetricEventWindow = 5 * time.Minute
|
||
|
||
// 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
|
||
// InboxStates 是 Weave 内部事件按状态的有界计数。
|
||
InboxStates map[string]uint64
|
||
// InboxOldestAgeSeconds 是待消费内部事件中最早 available_at 的年龄。
|
||
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
|
||
}
|
||
|
||
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"}
|
||
operationalInboxStates = []string{"PENDING", "LEASED", "PROCESSED", "QUARANTINED"}
|
||
operationalEventTypes = []string{"task.accepted", "task.started", "task.succeeded", "task.failed", "task.cancelled", "task.needs_review", "task.requeued", "workflow.started", "workflow.advanced", "workflow.recovered", "workflow.blocked", "workflow.stopped"}
|
||
phaseEventTypes = []string{"task.accepted", "task.started", "task.succeeded", "task.failed", "task.cancelled", "workflow.started", "workflow.stopped"}
|
||
)
|
||
|
||
// 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
|
||
}
|
||
inbox, inboxTruncated, err := boundedStateCounts(ctx, db, "weave_inbox", "ix_inbox_due", operationalInboxStates)
|
||
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
|
||
}
|
||
inboxAge, err := oldestAge(ctx, db, "weave_inbox", "ix_inbox_due", []string{"PENDING", "LEASED"}, "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
|
||
}
|
||
events, eventTruncated, err := boundedRecentEventCounts(ctx, db, now)
|
||
if err != nil {
|
||
return OperationalSnapshot{}, err
|
||
}
|
||
taskLatency, err := sampledTerminalLatency(ctx, db, "pulse_tasks", "ix_task_observe_terminal_completed", []string{"SUCCEEDED", "FAILED", "CANCELLED"}, now.Add(-operationalMetricEventWindow))
|
||
if err != nil {
|
||
return OperationalSnapshot{}, err
|
||
}
|
||
workflowLatency, err := sampledTerminalLatency(ctx, db, "weave_instances", "ix_workflow_observe_terminal_completed", []string{"SUCCEEDED", "FAILED", "CANCELLED"}, now.Add(-operationalMetricEventWindow))
|
||
if err != nil {
|
||
return OperationalSnapshot{}, err
|
||
}
|
||
phaseLatency, err := sampledPhaseLatency(ctx, db, now.Add(-operationalMetricEventWindow))
|
||
if err != nil {
|
||
return OperationalSnapshot{}, err
|
||
}
|
||
return OperationalSnapshot{
|
||
TaskStates: tasks, TaskOldestAgeSeconds: taskAge,
|
||
DeliveryStates: deliveries, DeliveryOldestAgeSeconds: deliveryAge,
|
||
WorkflowStates: workflows, WorkflowOldestAgeSeconds: workflowAge,
|
||
InboxStates: inbox, InboxOldestAgeSeconds: inboxAge,
|
||
RecentEvents: events,
|
||
TaskTerminalLatencySeconds: taskLatency.average, TaskTerminalLatencyMaxSeconds: taskLatency.maximum,
|
||
WorkflowTerminalLatencySeconds: workflowLatency.average, WorkflowTerminalLatencyMaxSeconds: workflowLatency.maximum,
|
||
TaskAcceptToStartLatencySeconds: phaseLatency.taskAcceptToStart.average, TaskAcceptToStartLatencyMaxSeconds: phaseLatency.taskAcceptToStart.maximum,
|
||
TaskAcceptToTerminalLatencySeconds: phaseLatency.taskAcceptToTerminal.average, TaskAcceptToTerminalLatencyMaxSeconds: phaseLatency.taskAcceptToTerminal.maximum,
|
||
WorkflowStartToTerminalLatencySeconds: phaseLatency.workflowStartToTerminal.average, WorkflowStartToTerminalLatencyMaxSeconds: phaseLatency.workflowStartToTerminal.maximum,
|
||
Truncated: taskTruncated || deliveryTruncated || workflowTruncated || inboxTruncated || eventTruncated || taskLatency.truncated || workflowLatency.truncated || phaseLatency.truncated,
|
||
}, 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 boundedRecentEventCounts(ctx context.Context, db *sql.DB, now time.Time) (map[string]uint64, bool, error) {
|
||
counts := make(map[string]uint64, len(operationalEventTypes))
|
||
truncated := false
|
||
cutoff := now.Add(-operationalMetricEventWindow)
|
||
for _, eventType := range operationalEventTypes {
|
||
query := "SELECT COUNT(*) FROM (SELECT id FROM orbit_events FORCE INDEX (ix_event_observe_type_occurred) WHERE type=? AND occurred_at>=? LIMIT ?) AS bounded"
|
||
var n uint64
|
||
if err := db.QueryRowContext(ctx, query, eventType, cutoff, operationalMetricSampleLimit+1).Scan(&n); err != nil {
|
||
return nil, false, err
|
||
}
|
||
if n > operationalMetricSampleLimit {
|
||
n = operationalMetricSampleLimit
|
||
truncated = true
|
||
}
|
||
counts[eventType] = n
|
||
}
|
||
return counts, truncated, nil
|
||
}
|
||
|
||
type terminalLatencySample struct {
|
||
average float64
|
||
maximum float64
|
||
truncated bool
|
||
}
|
||
|
||
type phaseLatencySamples struct {
|
||
taskAcceptToStart terminalLatencySample
|
||
taskAcceptToTerminal terminalLatencySample
|
||
workflowStartToTerminal terminalLatencySample
|
||
truncated bool
|
||
}
|
||
|
||
type phaseEvent struct {
|
||
kind string
|
||
id string
|
||
typeName string
|
||
at time.Time
|
||
}
|
||
|
||
// sampledPhaseLatency pairs only fixed event types in the recent bounded
|
||
// window. Missing history, retries and events outside the window are excluded;
|
||
// the result is a trend sample rather than an end-to-end SLO.
|
||
func sampledPhaseLatency(ctx context.Context, db *sql.DB, cutoff time.Time) (phaseLatencySamples, error) {
|
||
placeholders := "?"
|
||
args := make([]any, 0, len(phaseEventTypes)+2)
|
||
args = append(args, phaseEventTypes[0])
|
||
for _, eventType := range phaseEventTypes[1:] {
|
||
placeholders += ",?"
|
||
args = append(args, eventType)
|
||
}
|
||
args = append([]any{cutoff}, args...)
|
||
args = append(args, operationalMetricSampleLimit+1)
|
||
query := "SELECT aggregate_kind,aggregate_id,type,occurred_at FROM orbit_events FORCE INDEX (ix_event_observe_type_occurred) WHERE occurred_at>=? AND type IN (" + placeholders + ") LIMIT ?"
|
||
rows, err := db.QueryContext(ctx, query, args...)
|
||
if err != nil {
|
||
return phaseLatencySamples{}, err
|
||
}
|
||
defer rows.Close()
|
||
tasks := make(map[string][]phaseEvent)
|
||
workflows := make(map[string][]phaseEvent)
|
||
read := 0
|
||
for rows.Next() {
|
||
read++
|
||
var event phaseEvent
|
||
var id []byte
|
||
if err := rows.Scan(&event.kind, &id, &event.typeName, &event.at); err != nil {
|
||
return phaseLatencySamples{}, err
|
||
}
|
||
if read <= operationalMetricSampleLimit {
|
||
event.id = string(id)
|
||
switch event.kind {
|
||
case "task":
|
||
tasks[event.id] = append(tasks[event.id], event)
|
||
case "workflow":
|
||
workflows[event.id] = append(workflows[event.id], event)
|
||
}
|
||
}
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return phaseLatencySamples{}, err
|
||
}
|
||
result := phaseLatencySamples{truncated: read > operationalMetricSampleLimit}
|
||
result.taskAcceptToStart = pairPhaseLatencies(tasks, "task.accepted", "task.started")
|
||
result.taskAcceptToTerminal = pairPhaseLatencies(tasks, "task.accepted", "task.succeeded", "task.failed", "task.cancelled")
|
||
result.workflowStartToTerminal = pairPhaseLatencies(workflows, "workflow.started", "workflow.stopped")
|
||
result.taskAcceptToStart.truncated = false
|
||
result.taskAcceptToTerminal.truncated = false
|
||
result.workflowStartToTerminal.truncated = false
|
||
return result, nil
|
||
}
|
||
|
||
func pairPhaseLatencies(groups map[string][]phaseEvent, from string, to ...string) terminalLatencySample {
|
||
targets := make(map[string]struct{}, len(to))
|
||
for _, eventType := range to {
|
||
targets[eventType] = struct{}{}
|
||
}
|
||
var total, maximum float64
|
||
var count int
|
||
for _, events := range groups {
|
||
sort.Slice(events, func(i, j int) bool {
|
||
if events[i].at.Equal(events[j].at) {
|
||
return events[i].id < events[j].id
|
||
}
|
||
return events[i].at.Before(events[j].at)
|
||
})
|
||
var start *time.Time
|
||
var finish *time.Time
|
||
for _, event := range events {
|
||
if event.typeName == from && start == nil {
|
||
at := event.at.UTC()
|
||
start = &at
|
||
continue
|
||
}
|
||
if start != nil {
|
||
if _, ok := targets[event.typeName]; ok && !event.at.Before(*start) {
|
||
at := event.at.UTC()
|
||
finish = &at
|
||
break
|
||
}
|
||
}
|
||
}
|
||
if start == nil || finish == nil {
|
||
continue
|
||
}
|
||
seconds := finish.Sub(*start).Seconds()
|
||
if seconds < 0 {
|
||
seconds = 0
|
||
}
|
||
total += seconds
|
||
if seconds > maximum {
|
||
maximum = seconds
|
||
}
|
||
count++
|
||
}
|
||
result := terminalLatencySample{maximum: maximum}
|
||
if count > 0 {
|
||
result.average = total / float64(count)
|
||
}
|
||
return result
|
||
}
|
||
|
||
func sampledTerminalLatency(ctx context.Context, db *sql.DB, table, index string, states []string, cutoff time.Time) (terminalLatencySample, error) {
|
||
placeholders := "?"
|
||
args := make([]any, 0, len(states)+2)
|
||
args = append(args, states[0])
|
||
for _, state := range states[1:] {
|
||
placeholders += ",?"
|
||
args = append(args, state)
|
||
}
|
||
args = append(args, cutoff, operationalMetricSampleLimit+1)
|
||
query := "SELECT created_at,completed_at FROM " + table + " FORCE INDEX (" + index + ") WHERE state IN (" + placeholders + ") AND completed_at>=? ORDER BY completed_at,id LIMIT ?"
|
||
rows, err := db.QueryContext(ctx, query, args...)
|
||
if err != nil {
|
||
return terminalLatencySample{}, err
|
||
}
|
||
defer rows.Close()
|
||
var total, maximum float64
|
||
var read, valid int
|
||
for rows.Next() {
|
||
read++
|
||
if read > operationalMetricSampleLimit {
|
||
continue
|
||
}
|
||
var created, completed time.Time
|
||
if err := rows.Scan(&created, &completed); err != nil {
|
||
return terminalLatencySample{}, err
|
||
}
|
||
seconds := completed.Sub(created).Seconds()
|
||
if seconds < 0 {
|
||
seconds = 0
|
||
}
|
||
total += seconds
|
||
if seconds > maximum {
|
||
maximum = seconds
|
||
}
|
||
valid++
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return terminalLatencySample{}, err
|
||
}
|
||
result := terminalLatencySample{maximum: maximum, truncated: read > operationalMetricSampleLimit}
|
||
if valid > 0 {
|
||
result.average = total / float64(valid)
|
||
}
|
||
return result, 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
|
||
}
|