k2safe/orbit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: sample terminal business latency

Browse Source
This commit is contained in:
k2safe
2026-09-19 16:30:20 +08:00
parent 8ce9f4e1fa
commit 5a49dd5644
14 changed files with 139 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
internal/observability/business.go
View File

@@ -25,6 +25,14 @@ type BusinessSnapshot struct {
Truncated bool
// RecentEvents 固定事件类别在最近窗口内的有界数量。
RecentEvents map[string]uint64
// TaskTerminalLatencySeconds 最近窗口终态任务创建到完成的平均耗时样本。
TaskTerminalLatencySeconds float64
// TaskTerminalLatencyMaxSeconds 最近窗口终态任务样本最大耗时。
TaskTerminalLatencyMaxSeconds float64
// WorkflowTerminalLatencySeconds 最近窗口终态流程创建到完成的平均耗时样本。
WorkflowTerminalLatencySeconds float64
// WorkflowTerminalLatencyMaxSeconds 最近窗口终态流程样本最大耗时。
WorkflowTerminalLatencyMaxSeconds float64
}
// RecordBusinessSnapshot records one bounded database sample. A failed sample
@@ -58,6 +66,10 @@ func (r *Registry) RecordBusinessSnapshot(snapshot BusinessSnapshot, started tim
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_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)

4
internal/observability/catalog.go
View File

@@ -44,6 +44,10 @@ func catalog() map[string]descriptor {
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_events_recent"] = descriptor{gauge, "Bounded count of fixed event categories in the recent window", map[string][]string{"event": eventTypes}}
c["orbit_pulse_task_terminal_latency_seconds"] = descriptor{gauge, "Average sampled Pulse task create-to-terminal latency in seconds", nil}
c["orbit_pulse_task_terminal_latency_max_seconds"] = descriptor{gauge, "Maximum sampled Pulse task create-to-terminal latency in seconds", nil}
c["orbit_weave_workflow_terminal_latency_seconds"] = descriptor{gauge, "Average sampled Weave workflow create-to-terminal latency in seconds", nil}
c["orbit_weave_workflow_terminal_latency_max_seconds"] = descriptor{gauge, "Maximum sampled Weave workflow create-to-terminal latency in seconds", 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}

20
internal/observability/metrics_test.go
View File

@@ -117,14 +117,16 @@ 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,
RecentEvents: map[string]uint64{"task.accepted": 5},
Truncated: true,
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,
RecentEvents: map[string]uint64{"task.accepted": 5},
TaskTerminalLatencySeconds: 2.5, TaskTerminalLatencyMaxSeconds: 7.5,
WorkflowTerminalLatencySeconds: 3.5, WorkflowTerminalLatencyMaxSeconds: 9.5,
Truncated: true,
}, started, nil)
first := r.Render()
for _, want := range []string{
@@ -132,6 +134,8 @@ func TestBusinessSnapshotKeepsLastGoodStateOnFailure(t *testing.T) {
`orbit_result_deliveries_state{state="PENDING"} 2`,
`orbit_weave_workflows_state{state="RUNNING"} 1`,
`orbit_business_events_recent{event="task.accepted"} 5`,
"orbit_pulse_task_terminal_latency_seconds 2.5",
"orbit_weave_workflow_terminal_latency_max_seconds 9.5",
"orbit_pulse_tasks_oldest_age_seconds 12.5",
"orbit_business_snapshot_truncated 1",
`orbit_business_snapshot_total{outcome="success"} 1`,

2
internal/storage/mysql/application_resources_upgrade_integration_test.go
View File

@@ -69,6 +69,8 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) {
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(`ALTER TABLE orbit_events DROP INDEX ix_event_observe_type_occurred`)
f.exec(`ALTER TABLE pulse_tasks DROP INDEX ix_task_observe_terminal_completed`)
f.exec(`ALTER TABLE weave_instances DROP INDEX ix_workflow_observe_terminal_completed`)
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`)

7
internal/storage/mysql/migrations/000028_terminal_latency_metrics.up.sql Normal file
View File

@@ -0,0 +1,7 @@
-- 为终态任务/流程的最近完成耗时样本提供固定索引。
-- 采样只读已持久化时间，不修改状态、不引入外键或级联。
ALTER TABLE pulse_tasks
ADD KEY ix_task_observe_terminal_completed (state, completed_at, id);
ALTER TABLE weave_instances
ADD KEY ix_workflow_observe_terminal_completed (state, completed_at, id);

2
internal/storage/mysql/no_foreign_keys_integration_test.go
View File

@@ -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','ix_task_observe_state_available','ix_delivery_observe_state_available','ix_workflow_observe_state_updated','ix_event_observe_type_occurred') 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','ix_event_observe_type_occurred','ix_task_observe_terminal_completed','ix_workflow_observe_terminal_completed') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`)
if err != nil {
t.Fatal(err)
}

74
internal/storage/mysql/operational_metrics.go
View File

@@ -29,6 +29,14 @@ type OperationalSnapshot struct {
Truncated bool
// RecentEvents 是固定事件类别在最近窗口内的有界数量。
RecentEvents map[string]uint64
// TaskTerminalLatencySeconds 是最近窗口终态任务从创建到完成的平均耗时样本。
TaskTerminalLatencySeconds float64
// TaskTerminalLatencyMaxSeconds 是最近窗口终态任务样本的最大耗时。
TaskTerminalLatencyMaxSeconds float64
// WorkflowTerminalLatencySeconds 是最近窗口终态流程从创建到完成的平均耗时样本。
WorkflowTerminalLatencySeconds float64
// WorkflowTerminalLatencyMaxSeconds 是最近窗口终态流程样本的最大耗时。
WorkflowTerminalLatencyMaxSeconds float64
}
var (
@@ -76,12 +84,22 @@ func ReadOperationalSnapshot(ctx context.Context, db *sql.DB) (OperationalSnapsh
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
}
return OperationalSnapshot{
TaskStates: tasks, TaskOldestAgeSeconds: taskAge,
DeliveryStates: deliveries, DeliveryOldestAgeSeconds: deliveryAge,
WorkflowStates: workflows, WorkflowOldestAgeSeconds: workflowAge,
RecentEvents: events,
Truncated: taskTruncated || deliveryTruncated || workflowTruncated || eventTruncated,
RecentEvents: events,
TaskTerminalLatencySeconds: taskLatency.average, TaskTerminalLatencyMaxSeconds: taskLatency.maximum,
WorkflowTerminalLatencySeconds: workflowLatency.average, WorkflowTerminalLatencyMaxSeconds: workflowLatency.maximum,
Truncated: taskTruncated || deliveryTruncated || workflowTruncated || eventTruncated || taskLatency.truncated || workflowLatency.truncated,
}, nil
}
@@ -130,6 +148,58 @@ func boundedRecentEventCounts(ctx context.Context, db *sql.DB, now time.Time) (m
return counts, truncated, nil
}
type terminalLatencySample struct {
average float64
maximum float64
truncated bool
}
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))

13
internal/storage/mysql/operational_metrics_integration_test.go
View File

@@ -34,6 +34,9 @@ func TestIntegrationOperationalSnapshot(t *testing.T) {
{"pulse_tasks", "ix_task_observe_state_available"},
{"orbit_deliveries", "ix_delivery_observe_state_available"},
{"weave_instances", "ix_workflow_observe_state_updated"},
{"orbit_events", "ix_event_observe_type_occurred"},
{"pulse_tasks", "ix_task_observe_terminal_completed"},
{"weave_instances", "ix_workflow_observe_terminal_completed"},
} {
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 {
@@ -52,7 +55,7 @@ func TestIntegrationOperationalSnapshot(t *testing.T) {
}
f := newExecutionFixture(t, ctx, db)
f.submit(2)
task := f.submit(2)
snapshot, err := ReadOperationalSnapshot(ctx, db)
if err != nil {
t.Fatal(err)
@@ -77,6 +80,14 @@ func TestIntegrationOperationalSnapshot(t *testing.T) {
if _, ok := snapshot.WorkflowStates["RUNNING"]; !ok {
t.Fatal("missing running workflow state")
}
f.exec("UPDATE pulse_tasks SET state='SUCCEEDED', completed_at=UTC_TIMESTAMP(6), updated_at=UTC_TIMESTAMP(6), revision=revision+1 WHERE namespace_id=? AND id=?", f.resources.NamespaceID, task.ID)
completed, err := ReadOperationalSnapshot(ctx, db)
if err != nil {
t.Fatal(err)
}
if completed.TaskTerminalLatencySeconds < 0 || completed.TaskTerminalLatencyMaxSeconds < 0 {
t.Fatalf("negative terminal latency: %#v", completed)
}
cancelled, cancelNow := context.WithCancel(ctx)
cancelNow()