//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
}