feat: add bounded workflow execution measurement
This commit is contained in:
424
sdk/go/cmd/orbit-workflow-load/measure.go
Normal file
424
sdk/go/cmd/orbit-workflow-load/measure.go
Normal file
@@ -0,0 +1,424 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
|
||||
)
|
||||
|
||||
// workflowPayload 是两个并行分支共享的固定输入；不含凭据或外部地址。
|
||||
type workflowPayload struct {
|
||||
RunID string `json:"run_id"` // RunID 隔离本次测量，不作为业务幂等保证。
|
||||
Sequence int `json:"sequence"` // Sequence 从0开始的实例序号，用于发现错配和重复调用。
|
||||
}
|
||||
|
||||
func decodePayload(raw json.RawMessage) (workflowPayload, bool) {
|
||||
var wire struct {
|
||||
RunID *string `json:"run_id"`
|
||||
Sequence *int `json:"sequence"`
|
||||
}
|
||||
decoder := json.NewDecoder(bytes.NewReader(raw))
|
||||
decoder.DisallowUnknownFields()
|
||||
if decoder.Decode(&wire) != nil || wire.RunID == nil || wire.Sequence == nil {
|
||||
return workflowPayload{}, false
|
||||
}
|
||||
var trailing any
|
||||
if decoder.Decode(&trailing) != io.EOF {
|
||||
return workflowPayload{}, false
|
||||
}
|
||||
return workflowPayload{RunID: *wire.RunID, Sequence: *wire.Sequence}, true
|
||||
}
|
||||
|
||||
func handlerName(runID string) string { return "load.workflow." + runID }
|
||||
func definitionKey(runID string) string { return "orbit-workflow-load-" + runID }
|
||||
func requestKey(runID string, sequence int) string {
|
||||
return fmt.Sprintf("orbit-workflow-%s-%06d", runID, sequence)
|
||||
}
|
||||
func businessKey(runID string, sequence int) string {
|
||||
return fmt.Sprintf("orbit-workflow-%s-%06d", runID, sequence)
|
||||
}
|
||||
|
||||
type workflowAPI interface {
|
||||
GetQueue(context.Context, string) (orbit.Queue, error)
|
||||
CreateWorkflowDefinition(context.Context, string, orbit.CreateWorkflowDefinitionRequest) (orbit.WorkflowDefinitionResult, error)
|
||||
PublishWorkflowVersion(context.Context, string, string, orbit.PublishWorkflowVersionRequest) (orbit.WorkflowVersionResult, error)
|
||||
StartWorkflow(context.Context, string, orbit.StartWorkflowRequest) (orbit.WorkflowStartResult, error)
|
||||
GetWorkflow(context.Context, string) (orbit.WorkflowInstance, error)
|
||||
GetWorkflowSteps(context.Context, string) (orbit.WorkflowSteps, error)
|
||||
GetTask(context.Context, string) (orbit.Task, error)
|
||||
ListTaskAttempts(context.Context, string, orbit.HistoryPageRequest) (orbit.AttemptPage, error)
|
||||
GetTaskSource(context.Context, string) (orbit.TaskSource, error)
|
||||
}
|
||||
|
||||
type workflowAdmission struct {
|
||||
sequence int
|
||||
instance orbit.WorkflowInstance
|
||||
duration time.Duration
|
||||
code string
|
||||
}
|
||||
|
||||
type workflowVerification struct {
|
||||
sequence int
|
||||
duration time.Duration
|
||||
code string
|
||||
}
|
||||
|
||||
// invocationLog 有界保存每个流程的实际处理器调用，两个并行分支各应调用一次。
|
||||
type invocationLog struct {
|
||||
mu sync.Mutex
|
||||
calls []int
|
||||
invalid int
|
||||
}
|
||||
|
||||
func newInvocationLog(n int) *invocationLog { return &invocationLog{calls: make([]int, n)} }
|
||||
|
||||
func (log *invocationLog) handler(ctx context.Context, cfg config, runID string) orbit.WorkerHandler {
|
||||
return orbit.WorkerHandler{Capability: orbit.WorkerCapability{Handler: handlerName(runID), HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: func(work context.Context, execution orbit.WorkerExecution) (json.RawMessage, error) {
|
||||
payload, valid := decodePayload(execution.Task.Payload)
|
||||
if !valid || payload.RunID != runID || payload.Sequence < 0 || payload.Sequence >= cfg.Requests || execution.Task.Handler != handlerName(runID) || execution.Task.HandlerVersion != "v1" || execution.Task.PayloadVersion != "v1" || execution.Task.QueueID != cfg.QueueID {
|
||||
log.mu.Lock()
|
||||
log.invalid++
|
||||
log.mu.Unlock()
|
||||
return nil, &orbit.WorkerFailure{Code: "WORKFLOW_INPUT_MISMATCH"}
|
||||
}
|
||||
log.mu.Lock()
|
||||
log.calls[payload.Sequence]++
|
||||
log.mu.Unlock()
|
||||
select {
|
||||
case <-work.Done():
|
||||
return nil, work.Err()
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
default:
|
||||
}
|
||||
return execution.Task.Payload, nil
|
||||
}}
|
||||
}
|
||||
|
||||
func workflowSpec(queue, handler string, timeout time.Duration) orbit.WorkflowSpecification {
|
||||
typ := orbit.WorkflowType{Kind: "OBJECT", Fields: map[string]orbit.WorkflowType{"run_id": {Kind: "STRING"}, "sequence": {Kind: "NUMBER"}}}
|
||||
task := orbit.WorkflowTask{QueueID: queue, Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1", InputType: typ, Input: orbit.WorkflowRef("input"), ResultType: typ, TimeoutMS: uint64(timeout.Milliseconds()), Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: uint64(timeout.Milliseconds())}, RecoverySafety: orbit.RecoveryReconcileOnly}
|
||||
parallel := orbit.WorkflowParallel{Branches: []orbit.WorkflowParallelBranch{
|
||||
{Name: "left", Body: orbit.WorkflowBlock{Steps: []orbit.WorkflowStep{{Key: "left", Kind: "TASK", Task: &task}}, OutputType: typ, Output: orbit.WorkflowRef("left")}},
|
||||
{Name: "right", Body: orbit.WorkflowBlock{Steps: []orbit.WorkflowStep{{Key: "right", Kind: "TASK", Task: &task}}, OutputType: typ, Output: orbit.WorkflowRef("right")}},
|
||||
}}
|
||||
output := orbit.WorkflowType{Kind: "OBJECT", Fields: map[string]orbit.WorkflowType{"left": typ, "right": typ}}
|
||||
return orbit.WorkflowSpecification{SchemaVersion: 1, InputType: typ, TimeoutMS: uint64(timeout.Milliseconds()), FailurePolicy: "REVIEW", Body: orbit.WorkflowBlock{Steps: []orbit.WorkflowStep{{Key: "pair", Kind: "PARALLEL", Parallel: ¶llel}}, OutputType: output, Output: orbit.WorkflowRef("pair")}}
|
||||
}
|
||||
|
||||
func measure(parent context.Context, cfg config, api *orbit.Client, runID string) workflowReport {
|
||||
return measureClient(parent, cfg, api, runID)
|
||||
}
|
||||
|
||||
func measureClient(parent context.Context, cfg config, api *orbit.Client, runID string) workflowReport {
|
||||
r := newReport(cfg, runID)
|
||||
ctx, cancel := context.WithTimeout(parent, cfg.RunTimeout)
|
||||
defer cancel()
|
||||
requestCtx, stop := context.WithTimeout(ctx, cfg.Timeout)
|
||||
queue, err := api.GetQueue(requestCtx, cfg.QueueID)
|
||||
stop()
|
||||
if err != nil {
|
||||
r.Errors["PREFLIGHT_"+classify(err)]++
|
||||
return r.finish()
|
||||
}
|
||||
r.Queue = &queue
|
||||
if queue.ID != cfg.QueueID || queue.Paused || queue.MaxConcurrency == 0 || queue.RatePerSecond == 0 {
|
||||
r.Errors["QUEUE_UNAVAILABLE"]++
|
||||
return r.finish()
|
||||
}
|
||||
handler := handlerName(runID)
|
||||
typicalTimeout := cfg.RunTimeout
|
||||
if typicalTimeout > 30*time.Second {
|
||||
typicalTimeout = 30 * time.Second
|
||||
}
|
||||
spec := workflowSpec(cfg.QueueID, handler, typicalTimeout)
|
||||
requestCtx, stop = context.WithTimeout(ctx, cfg.Timeout)
|
||||
definition, err := api.CreateWorkflowDefinition(requestCtx, definitionKey(runID), orbit.CreateWorkflowDefinitionRequest{Name: definitionKey(runID), Description: "bounded parallel workflow verification"})
|
||||
stop()
|
||||
if err != nil {
|
||||
r.Errors["CREATE_"+classify(err)]++
|
||||
return r.finish()
|
||||
}
|
||||
r.DefinitionID = definition.Definition.ID
|
||||
if !validID(r.DefinitionID) {
|
||||
r.Errors["INVALID_DEFINITION_RECEIPT"]++
|
||||
return r.finish()
|
||||
}
|
||||
requestCtx, stop = context.WithTimeout(ctx, cfg.Timeout)
|
||||
version, err := api.PublishWorkflowVersion(requestCtx, r.DefinitionID, runID, orbit.PublishWorkflowVersionRequest{Version: "v1", Specification: spec, Reason: "bounded parallel workflow verification"})
|
||||
stop()
|
||||
if err != nil {
|
||||
r.Errors["PUBLISH_"+classify(err)]++
|
||||
return r.finish()
|
||||
}
|
||||
r.VersionID = version.Version.ID
|
||||
if !validID(r.VersionID) || version.Version.Version != "v1" || version.Version.DefinitionID != r.DefinitionID {
|
||||
r.Errors["INVALID_VERSION_RECEIPT"]++
|
||||
return r.finish()
|
||||
}
|
||||
|
||||
log := newInvocationLog(cfg.Requests)
|
||||
workers := make([]*orbit.Worker, 0, cfg.Workers)
|
||||
for i := 0; i < cfg.Workers; i++ {
|
||||
worker, e := orbit.NewWorker(api, orbit.WorkerOptions{QueueID: cfg.QueueID, Capacity: uint32(cfg.Capacity), ShutdownTimeout: 5 * time.Second, Handlers: []orbit.WorkerHandler{log.handler(ctx, cfg, runID)}})
|
||||
if e != nil {
|
||||
r.Errors["WORKER_CONFIG"]++
|
||||
return r.finish()
|
||||
}
|
||||
workers = append(workers, worker)
|
||||
}
|
||||
workerCtx, stopWorkers := context.WithCancel(ctx)
|
||||
done := make(chan error, len(workers))
|
||||
for _, worker := range workers {
|
||||
go func(w *orbit.Worker) {
|
||||
e := w.Run(workerCtx)
|
||||
if workerCtx.Err() == nil {
|
||||
cancel()
|
||||
}
|
||||
done <- e
|
||||
}(worker)
|
||||
}
|
||||
|
||||
admissions := submitWorkflows(ctx, cfg, api, runID, r.DefinitionID)
|
||||
r.Attempted = len(admissions)
|
||||
r.NotSubmitted = cfg.Requests - r.Attempted
|
||||
var accepted []workflowAdmission
|
||||
var admissionDurations []time.Duration
|
||||
for _, a := range admissions {
|
||||
admissionDurations = append(admissionDurations, a.duration)
|
||||
if a.code != "" {
|
||||
r.AdmissionErrors++
|
||||
r.Errors["ADMISSION_"+a.code]++
|
||||
continue
|
||||
}
|
||||
accepted = append(accepted, a)
|
||||
}
|
||||
r.Accepted = len(accepted)
|
||||
r.Admission = distributionOf(admissionDurations)
|
||||
r.AdmissionFinishedAt = time.Now().UTC()
|
||||
verified := verifyWorkflows(ctx, cfg, api, runID, r.DefinitionID, accepted)
|
||||
var completions []time.Duration
|
||||
for _, v := range verified {
|
||||
if v.code != "" {
|
||||
r.Errors[v.code]++
|
||||
continue
|
||||
}
|
||||
r.VerifiedSuccess++
|
||||
completions = append(completions, v.duration)
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
r.Errors["RUN_"+classify(ctx.Err())]++
|
||||
}
|
||||
stopWorkers()
|
||||
for range workers {
|
||||
if e := <-done; e != nil {
|
||||
r.Errors["WORKER_STOP"]++
|
||||
}
|
||||
}
|
||||
for _, worker := range workers {
|
||||
stats := worker.Stats()
|
||||
r.WorkerStarted += stats.Started
|
||||
r.Reconnects += stats.Reconnects
|
||||
r.Renewals += stats.RenewalsConfirmed
|
||||
}
|
||||
log.mu.Lock()
|
||||
for _, calls := range log.calls {
|
||||
r.HandlerCalls += calls
|
||||
if calls > 2 {
|
||||
r.DuplicateInvocations += calls - 2
|
||||
}
|
||||
if calls != 2 {
|
||||
r.Errors["HANDLER_CALL_MISMATCH"]++
|
||||
}
|
||||
}
|
||||
r.Errors["INVALID_HANDLER_INPUT"] += log.invalid
|
||||
log.mu.Unlock()
|
||||
r.Completion = distributionOf(completions)
|
||||
return r.finish()
|
||||
}
|
||||
|
||||
func submitWorkflows(ctx context.Context, cfg config, api workflowAPI, runID, definitionID string) []workflowAdmission {
|
||||
jobs := make(chan int)
|
||||
out := make(chan workflowAdmission, cfg.Requests)
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < cfg.Concurrency; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for sequence := range jobs {
|
||||
if ctx.Err() != nil {
|
||||
continue
|
||||
}
|
||||
payload, _ := json.Marshal(workflowPayload{RunID: runID, Sequence: sequence})
|
||||
started := time.Now()
|
||||
call, stop := context.WithTimeout(ctx, cfg.Timeout)
|
||||
result, err := api.StartWorkflow(call, requestKey(runID, sequence), orbit.StartWorkflowRequest{DefinitionID: definitionID, Version: "v1", Input: payload, BusinessKey: businessKey(runID, sequence)})
|
||||
stop()
|
||||
a := workflowAdmission{sequence: sequence, instance: result.Instance, duration: time.Since(started)}
|
||||
if err != nil {
|
||||
a.code = classify(err)
|
||||
} else if result.Deduplicated || !validID(result.Instance.ID) || result.Instance.DefinitionID != definitionID || result.Instance.Version != "v1" || result.Instance.BusinessKey != businessKey(runID, sequence) {
|
||||
a.code = "INVALID_RECEIPT"
|
||||
}
|
||||
out <- a
|
||||
}
|
||||
}()
|
||||
}
|
||||
feed:
|
||||
for sequence := 0; sequence < cfg.Requests; sequence++ {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
break feed
|
||||
case jobs <- sequence:
|
||||
}
|
||||
}
|
||||
close(jobs)
|
||||
wg.Wait()
|
||||
close(out)
|
||||
results := make([]workflowAdmission, 0, cfg.Requests)
|
||||
for result := range out {
|
||||
results = append(results, result)
|
||||
}
|
||||
return results
|
||||
}
|
||||
|
||||
func verifyWorkflows(ctx context.Context, cfg config, api workflowAPI, runID, definitionID string, admissions []workflowAdmission) []workflowVerification {
|
||||
jobs := make(chan workflowAdmission, len(admissions))
|
||||
for _, admission := range admissions {
|
||||
jobs <- admission
|
||||
}
|
||||
close(jobs)
|
||||
out := make(chan workflowVerification, len(admissions))
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < cfg.Concurrency; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for admission := range jobs {
|
||||
duration, code := verifyWorkflow(ctx, cfg, api, runID, definitionID, admission)
|
||||
out <- workflowVerification{sequence: admission.sequence, duration: duration, code: code}
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
close(out)
|
||||
results := make([]workflowVerification, 0, len(admissions))
|
||||
for result := range out {
|
||||
results = append(results, result)
|
||||
}
|
||||
return results
|
||||
}
|
||||
|
||||
func verifyWorkflow(ctx context.Context, cfg config, api workflowAPI, runID, definitionID string, admission workflowAdmission) (time.Duration, string) {
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return 0, "VERIFY_" + classify(ctx.Err())
|
||||
}
|
||||
call, stop := context.WithTimeout(ctx, cfg.Timeout)
|
||||
instance, err := api.GetWorkflow(call, admission.instance.ID)
|
||||
stop()
|
||||
if err != nil {
|
||||
return 0, "READ_" + classify(err)
|
||||
}
|
||||
if instance.ID != admission.instance.ID || instance.DefinitionID != definitionID || instance.Version != "v1" || instance.BusinessKey != businessKey(runID, admission.sequence) {
|
||||
return 0, "WORKFLOW_MISMATCH"
|
||||
}
|
||||
switch instance.State {
|
||||
case "SUCCEEDED":
|
||||
if instance.CompletedAt == nil || instance.CreatedAt.IsZero() || instance.CompletedAt.Before(instance.CreatedAt) {
|
||||
return 0, "COMPLETION_MISMATCH"
|
||||
}
|
||||
call, stop = context.WithTimeout(ctx, cfg.Timeout)
|
||||
steps, e := api.GetWorkflowSteps(call, instance.ID)
|
||||
stop()
|
||||
if e != nil {
|
||||
return 0, "STEPS_" + classify(e)
|
||||
}
|
||||
if code := verifySteps(ctx, cfg, api, instance, runID, admission.sequence, steps); code != "" {
|
||||
return 0, code
|
||||
}
|
||||
if !matchesOutput(instance.Output, runID, admission.sequence) {
|
||||
return 0, "OUTPUT_MISMATCH"
|
||||
}
|
||||
return instance.CompletedAt.Sub(instance.CreatedAt), ""
|
||||
case "NEEDS_REVIEW", "FAILED", "CANCELLED":
|
||||
return 0, "WORKFLOW_" + instance.State
|
||||
case "RUNNING", "WAITING", "PAUSED", "PENDING":
|
||||
default:
|
||||
return 0, "UNKNOWN_WORKFLOW_STATE"
|
||||
}
|
||||
timer := time.NewTimer(cfg.PollInterval)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
timer.Stop()
|
||||
return 0, "VERIFY_" + classify(ctx.Err())
|
||||
case <-timer.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func verifySteps(ctx context.Context, cfg config, api workflowAPI, instance orbit.WorkflowInstance, runID string, sequence int, steps orbit.WorkflowSteps) string {
|
||||
var taskSteps []orbit.WorkflowStepStatus
|
||||
for _, step := range steps.Steps {
|
||||
if step.Kind == "TASK" {
|
||||
taskSteps = append(taskSteps, step)
|
||||
}
|
||||
}
|
||||
if len(taskSteps) != 2 {
|
||||
return "STEP_COUNT_MISMATCH"
|
||||
}
|
||||
for _, step := range taskSteps {
|
||||
if step.State != "SUCCEEDED" || step.TaskID == nil || !validID(*step.TaskID) || step.ActivationSequence != 1 || step.ActivationID == nil {
|
||||
return "STEP_MISMATCH"
|
||||
}
|
||||
call, stop := context.WithTimeout(ctx, cfg.Timeout)
|
||||
task, err := api.GetTask(call, *step.TaskID)
|
||||
stop()
|
||||
if err != nil {
|
||||
return "TASK_" + classify(err)
|
||||
}
|
||||
payload, valid := decodePayload(task.Result)
|
||||
if task.ID != *step.TaskID || task.State != orbit.TaskSucceeded || task.EffectUncertain || task.AttemptCount != 1 || !valid || payload != (workflowPayload{RunID: runID, Sequence: sequence}) {
|
||||
return "TASK_MISMATCH"
|
||||
}
|
||||
call, stop = context.WithTimeout(ctx, cfg.Timeout)
|
||||
history, err := api.ListTaskAttempts(call, task.ID, orbit.HistoryPageRequest{Limit: 2})
|
||||
stop()
|
||||
if err != nil {
|
||||
return "HISTORY_" + classify(err)
|
||||
}
|
||||
if len(history.Items) != 1 || history.NextCursor != "" || history.Items[0].TaskID != task.ID || history.Items[0].State != orbit.AttemptSucceeded {
|
||||
return "ATTEMPT_MISMATCH"
|
||||
}
|
||||
call, stop = context.WithTimeout(ctx, cfg.Timeout)
|
||||
source, err := api.GetTaskSource(call, task.ID)
|
||||
stop()
|
||||
if err != nil || source.Kind != "WORKFLOW" || source.Workflow == nil || source.Workflow.WorkflowID != instance.ID || source.Workflow.DefinitionID != instance.DefinitionID || source.Workflow.DefinitionVersionID != instance.DefinitionVersionID || source.Workflow.Version != instance.Version || source.Workflow.StepID != step.ID || source.Workflow.ActivationID != *step.ActivationID || source.Workflow.ActivationSequence != 1 {
|
||||
return "SOURCE_MISMATCH"
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func matchesOutput(raw json.RawMessage, runID string, sequence int) bool {
|
||||
var wire struct {
|
||||
Left workflowPayload `json:"left"`
|
||||
Right workflowPayload `json:"right"`
|
||||
}
|
||||
decoder := json.NewDecoder(bytes.NewReader(raw))
|
||||
decoder.DisallowUnknownFields()
|
||||
if decoder.Decode(&wire) != nil {
|
||||
return false
|
||||
}
|
||||
var trailing any
|
||||
if decoder.Decode(&trailing) != io.EOF {
|
||||
return false
|
||||
}
|
||||
return wire.Left == (workflowPayload{RunID: runID, Sequence: sequence}) && wire.Right == wire.Left
|
||||
}
|
||||
Reference in New Issue
Block a user