//go:build torture package spindle import ( "bytes" "context" "errors" "fmt" "io" "strconv" "strings" "sync" "sync/atomic" "time" "gopkg.in/yaml.v3" "tangled.org/core/spindle/engine" "tangled.org/core/spindle/models" "tangled.org/core/spindle/quota" "tangled.org/core/spindle/secrets" "tangled.org/core/workflow" ) // no real work runs here, which means the numbers only measure bookkeeping type tortureEngine struct { slots *engine.SemaphoreSlotter running atomic.Int64 started atomic.Int64 mu sync.Mutex order []string took map[string]time.Duration } func newTortureEngine() *tortureEngine { return &tortureEngine{slots: engine.NewSemaphoreSlotter(0), took: map[string]time.Duration{}} } type tortureStep struct { name string command string } func (s tortureStep) Name() string { return s.name } func (s tortureStep) Command() string { return s.command } func (s tortureStep) Kind() models.StepKind { return models.StepKindUser } func (e *tortureEngine) InitWorkflow(twf workflow.WorkflowSpec, tpl workflow.PipelineSpec) (*models.Workflow, error) { var spec struct { Steps []struct { Name string `yaml:"name"` Command string `yaml:"command"` } `yaml:"steps"` } if err := yaml.Unmarshal([]byte(twf.Raw), &spec); err != nil { return nil, err } md := tpl.TriggerMetadata if md == nil || md.Repo == nil || md.Repo.Did == "" { return nil, errors.New("missing owner in trigger metadata") } wf := &models.Workflow{Name: twf.Name, OwnerDID: md.Repo.Did, RepoDID: md.Repo.Did} if md.Repo.RepoDid != nil { wf.RepoDID = *md.Repo.RepoDid } for _, s := range spec.Steps { wf.Steps = append(wf.Steps, tortureStep{name: s.Name, command: s.Command}) } return wf, nil } func (e *tortureEngine) SetupWorkflow(_ context.Context, _ models.WorkflowId, wf *models.Workflow, _ models.WorkflowLogger) error { e.started.Add(1) e.running.Add(1) e.mu.Lock() e.order = append(e.order, wf.OwnerDID) e.mu.Unlock() return nil } func (e *tortureEngine) WorkflowTimeout() time.Duration { return 5 * time.Minute } func (e *tortureEngine) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, mode engine.AcquireMode) (engine.WorkflowSlot, error) { return e.slots.AcquireWorkflowSlot(ctx, wid, wf, mode) } func (e *tortureEngine) DestroyWorkflow(context.Context, models.WorkflowId) error { e.running.Add(-1) return nil } func (e *tortureEngine) QuotaResources(*models.Workflow) quota.Resources { return quota.Resources{quota.ResourceWorkflows: 1} } func (e *tortureEngine) RunStep(ctx context.Context, _ models.WorkflowId, w *models.Workflow, idx int, _ []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error { out := wfLogger.DataWriter(idx, "stdout") start := time.Now() defer func() { e.mu.Lock() e.took[w.Steps[idx].Command()] = time.Since(start) e.mu.Unlock() }() for _, command := range strings.Split(w.Steps[idx].Command(), ";") { verb, arg, _ := strings.Cut(strings.TrimSpace(command), " ") n, numErr := strconv.Atoi(arg) switch verb { case "sleep": d, err := time.ParseDuration(arg) if err != nil { return err } if err := sleepCtx(ctx, d); err != nil { return err } case "log", "lines", "raw", "dribble", "longline": if numErr != nil { return numErr } if err := writeOutput(out, verb, n); err != nil { return err } case "fail": return errors.New("torture step failed on purpose") case "hang": <-ctx.Done() return ctx.Err() case "": default: return fmt.Errorf("unknown torture command %q", verb) } } return nil } func (e *tortureEngine) startOrder() []string { e.mu.Lock() defer e.mu.Unlock() return append([]string(nil), e.order...) } func (e *tortureEngine) tookFor(command string) time.Duration { e.mu.Lock() defer e.mu.Unlock() return e.took[command] } func writeOutput(out io.Writer, verb string, n int) error { switch verb { case "log": chunk := bytes.Repeat([]byte("x"), 4095) chunk = append(chunk, '\n') for n > 0 { if _, err := out.Write(chunk[:min(n, len(chunk))]); err != nil { return err } n -= len(chunk) } return nil case "lines": for i := range n { if _, err := fmt.Fprintf(out, "line %d\n", i); err != nil { return err } } return nil } chunk := map[string]int{"raw": 64 << 10, "dribble": 16, "longline": 64 << 10}[verb] return writeChunks(out, n, chunk, verb == "longline") } func writeChunks(out io.Writer, total, chunk int, newline bool) error { buf := bytes.Repeat([]byte("x"), chunk) for total > 0 { k := min(total, chunk) if _, err := out.Write(buf[:k]); err != nil { return err } total -= k } if newline { _, err := out.Write([]byte("\n")) return err } return nil }