feat: add bounded business state observability

2026-09-18 20:44:41 +08:00
parent bea8efa790
commit 5284206731
18 changed files with 450 additions and 12 deletions
--- 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`)
--- a/internal/storage/mysql/migrations/000026_operational_metrics.up.sql
+++ 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);
--- 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)
 		}
--- a/internal/storage/mysql/operational_metrics.go
+++ 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
+}
--- a/internal/storage/mysql/operational_metrics_integration_test.go
+++ 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")
+	}
+}