orbit/internal/storage/mysql/operational_metrics.go

package mysql

import (
	"context"
	"database/sql"
	"errors"
	"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
	// Truncated 表示至少一个状态达到有界采样上限。
	Truncated bool
	// RecentEvents 是固定事件类别在最近窗口内的有界数量。
	RecentEvents map[string]uint64
}

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"}
	operationalEventTypes     = []string{"task.accepted", "task.started", "task.succeeded", "task.failed", "task.needs_review", "task.requeued", "workflow.started", "workflow.advanced", "workflow.recovered", "workflow.blocked", "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
	}
	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
	}
	events, eventTruncated, err := boundedRecentEventCounts(ctx, db, now)
	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,
	}, 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
}

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
}