feat: sample workflow phase latency
This commit is contained in:
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"sort"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -41,6 +42,18 @@ type OperationalSnapshot struct {
|
||||
WorkflowTerminalLatencySeconds float64
|
||||
// WorkflowTerminalLatencyMaxSeconds 是最近窗口终态流程样本的最大耗时。
|
||||
WorkflowTerminalLatencyMaxSeconds float64
|
||||
// TaskAcceptToStartLatencySeconds 是最近窗口任务从受理事件到开始事件的平均耗时样本。
|
||||
TaskAcceptToStartLatencySeconds float64
|
||||
// TaskAcceptToStartLatencyMaxSeconds 是最近窗口任务受理到开始耗时样本的最大值。
|
||||
TaskAcceptToStartLatencyMaxSeconds float64
|
||||
// TaskAcceptToTerminalLatencySeconds 是最近窗口任务从受理到确定终态事件的平均耗时样本。
|
||||
TaskAcceptToTerminalLatencySeconds float64
|
||||
// TaskAcceptToTerminalLatencyMaxSeconds 是最近窗口任务受理到确定终态耗时样本的最大值。
|
||||
TaskAcceptToTerminalLatencyMaxSeconds float64
|
||||
// WorkflowStartToTerminalLatencySeconds 是最近窗口流程从开始到停止事件的平均耗时样本。
|
||||
WorkflowStartToTerminalLatencySeconds float64
|
||||
// WorkflowStartToTerminalLatencyMaxSeconds 是最近窗口流程开始到停止耗时样本的最大值。
|
||||
WorkflowStartToTerminalLatencyMaxSeconds float64
|
||||
}
|
||||
|
||||
var (
|
||||
@@ -48,7 +61,8 @@ var (
|
||||
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.needs_review", "task.requeued", "workflow.started", "workflow.advanced", "workflow.recovered", "workflow.blocked", "workflow.stopped"}
|
||||
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
|
||||
@@ -105,6 +119,10 @@ func ReadOperationalSnapshot(ctx context.Context, db *sql.DB) (OperationalSnapsh
|
||||
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,
|
||||
@@ -113,7 +131,10 @@ func ReadOperationalSnapshot(ctx context.Context, db *sql.DB) (OperationalSnapsh
|
||||
RecentEvents: events,
|
||||
TaskTerminalLatencySeconds: taskLatency.average, TaskTerminalLatencyMaxSeconds: taskLatency.maximum,
|
||||
WorkflowTerminalLatencySeconds: workflowLatency.average, WorkflowTerminalLatencyMaxSeconds: workflowLatency.maximum,
|
||||
Truncated: taskTruncated || deliveryTruncated || workflowTruncated || inboxTruncated || eventTruncated || taskLatency.truncated || workflowLatency.truncated,
|
||||
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
|
||||
}
|
||||
|
||||
@@ -168,6 +189,122 @@ type terminalLatencySample struct {
|
||||
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)
|
||||
|
||||
@@ -90,6 +90,12 @@ func TestIntegrationOperationalSnapshot(t *testing.T) {
|
||||
}
|
||||
}
|
||||
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)
|
||||
for _, event := range []struct {
|
||||
revision uint64
|
||||
typeName string
|
||||
}{{2, "task.started"}, {3, "task.succeeded"}} {
|
||||
f.exec("INSERT INTO orbit_events(namespace_id,id,aggregate_kind,aggregate_id,aggregate_revision,type,payload,occurred_at) VALUES (?,?,'task',?,?,?,'{}',UTC_TIMESTAMP(6))", f.resources.NamespaceID, executionID(t), task.ID, event.revision, event.typeName)
|
||||
}
|
||||
completed, err := ReadOperationalSnapshot(ctx, db)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -97,6 +103,9 @@ func TestIntegrationOperationalSnapshot(t *testing.T) {
|
||||
if completed.TaskTerminalLatencySeconds < 0 || completed.TaskTerminalLatencyMaxSeconds < 0 {
|
||||
t.Fatalf("negative terminal latency: %#v", completed)
|
||||
}
|
||||
if completed.TaskAcceptToStartLatencySeconds < 0 || completed.TaskAcceptToTerminalLatencySeconds < 0 {
|
||||
t.Fatalf("negative task phase latency: %#v", completed)
|
||||
}
|
||||
|
||||
cancelled, cancelNow := context.WithCancel(ctx)
|
||||
cancelNow()
|
||||
|
||||
Reference in New Issue
Block a user