//go:build torture
package spindle
import (
"context"
"database/sql"
"fmt"
"io/fs"
"math"
"net"
"net/url"
"os"
"path/filepath"
"runtime"
"runtime/debug"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"github.com/gorilla/websocket"
"tangled.org/core/api/org_tangled"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/engine"
"tangled.org/core/spindle/models"
pipelinecodec "tangled.org/core/spindle/pipeline"
"tangled.org/core/workflow"
)
var tqSeq atomic.Int64
func tqWorld(r *tortureRun, owners, perOwner, workers int) (*tortureWorld, []*tortureRepo) {
w := newTortureWorld(r)
w.maxJobs = workers
k := w.addKnot()
repos := w.addRepos(owners*perOwner, func(i int) (*tortureKnot, syntax.DID) {
return k, ownerFor(i, perOwner)
}, 0, nil)
w.boot()
w.seed(repos, 3*time.Minute)
return w, repos
}
func (w *tortureWorld) tqPipeline(repo *tortureRepo, workflows int, steps ...string) (models.PipelineId, workflow.PipelineSpec, *models.PipelineRecord, error) {
n := tqSeq.Add(1)
id := models.PipelineId(fmt.Sprintf("tq%09d", n))
repoDid := repo.did.String()
rkey := repo.name
actor := repo.owner.String()
tpl := workflow.PipelineSpec{
TriggerMetadata: &workflow.TriggerMetadata{
Kind: "push",
Actor: &actor,
Push: &workflow.PushTriggerData{NewSha: fmt.Sprintf("%040x", n), OldSha: zeroSha, Ref: "refs/heads/main"},
Repo: &workflow.TriggerRepo{
Did: repo.owner.String(), Knot: repo.knot.host(), Repo: &rkey, RepoDid: &repoDid, DefaultBranch: "main",
},
},
}
for i := range workflows {
name := "ci.yaml"
if workflows > 1 {
name = fmt.Sprintf("w%d.yaml", i)
}
tpl.Workflows = append(tpl.Workflows, &workflow.WorkflowSpec{
Name: name, Engine: "torture", Raw: tortureWorkflow(steps...), Clone: &workflow.CloneSpec{Skip: true},
})
}
record, err := pipelinecodec.FromTangled(id, time.Now(), tpl)
return id, tpl, record, err
}
func (w *tortureWorld) tqEnqueueTimed(lat *latencies, repo *tortureRepo, workflows int, steps ...string) (models.PipelineId, error) {
id, tpl, record, err := w.tqPipeline(repo, workflows, steps...)
if err != nil {
return "", err
}
start := time.Now()
err = w.s.processPipeline(context.Background(), repo.did, tpl, record, nil, false)
if lat != nil {
lat.since(start)
}
return id, err
}
func (w *tortureWorld) tqEnqueue(repo *tortureRepo, workflows int, steps ...string) (models.PipelineId, error) {
return w.tqEnqueueTimed(nil, repo, workflows, steps...)
}
func (w *tortureWorld) tqEnqueueMany(repos []*tortureRepo, n, par, workflows int, lat *latencies, steps ...string) []models.PipelineId {
ids := make([]models.PipelineId, n)
var next atomic.Int64
var wg sync.WaitGroup
for range par {
wg.Go(func() {
for {
i := int(next.Add(1)) - 1
if i >= n {
return
}
id, err := w.tqEnqueueTimed(lat, repos[i%len(repos)], workflows, steps...)
if err != nil {
w.r.count("enqueue_error", 1)
continue
}
ids[i] = id
}
})
}
wg.Wait()
return ids
}
func (w *tortureWorld) tqFinished(within time.Duration) bool {
return w.r.waitFor(within, func() bool { return w.unfinishedWorkflows() == 0 })
}
func (w *tortureWorld) tqWorkflowDone(id models.PipelineId, name string) bool {
st, err := w.s.db.GetStatus(models.WorkflowId{PipelineId: id, Name: name})
return err == nil && st.IsFinish()
}
func (w *tortureWorld) tqStartDelays(ids []models.PipelineId, name string, lat *latencies) {
for _, id := range ids {
if id == "" {
continue
}
d, ok, err := w.s.db.WorkflowStartupDelay(context.Background(), models.WorkflowId{PipelineId: id, Name: name})
if err == nil && ok {
lat.add(d)
}
}
}
// must match logWriteTimeout in xrpc/ci_pipeline_subscribe_logs.go
const tqLogWriteTimeout = time.Minute
func tqPeaks(r *tortureRun, fn func(), gauges ...string) []float64 {
peaks := make([]float64, len(gauges))
for i, g := range gauges {
peaks[i] = r.gaugeValue(g)
}
done := make(chan struct{})
var wg sync.WaitGroup
wg.Go(func() {
tick := time.NewTicker(25 * time.Millisecond)
defer tick.Stop()
for {
select {
case <-done:
return
case <-tick.C:
for i, g := range gauges {
peaks[i] = max(peaks[i], r.gaugeValue(g))
}
}
}
})
fn()
close(done)
wg.Wait()
return peaks
}
func tqSettle(r *tortureRun, phase, gauge string, ceiling float64, within time.Duration) {
ok := r.waitFor(within, func() bool { return r.gaugeValue(gauge) <= ceiling })
r.check(phase+": "+gauge+" settles", ok, "%s is %.0f after %v, want <= %.0f", gauge, r.gaugeValue(gauge), within, ceiling)
if ok {
return
}
switch gauge {
case "goroutines":
r.dumpGoroutines()
case "fds":
tqFdSummary(r)
}
}
func tqSettleBoth(r *tortureRun, phase string, baseGoroutines, baseFds, fdSlack float64, within time.Duration) {
tqSettle(r, phase, "goroutines", baseGoroutines+30, within)
tqSettle(r, phase, "fds", baseFds+fdSlack, within)
}
func tqFdSummary(r *tortureRun) {
targets, err := fdTargets()
if err != nil {
return
}
kinds := map[string]int{}
others := map[string]int{}
for _, target := range targets {
switch {
case target == "":
kinds["gone"]++
case strings.HasPrefix(target, "socket:"):
kinds["socket"]++
case strings.HasPrefix(target, "pipe:"):
kinds["pipe"]++
case strings.HasPrefix(target, "anon_inode:"):
kinds[target]++
case strings.HasSuffix(target, ".log"):
kinds["workflow log"]++
default:
kinds["other"]++
others[filepath.Base(target)]++
}
}
r.note("fds by kind: %v", kinds)
r.note("fds of kind other by file name: %v", others)
}
func tqLargestFileMB(root string) float64 {
var largest int64
_ = filepath.WalkDir(root, func(_ string, d fs.DirEntry, err error) error {
if err != nil || d.IsDir() {
return nil
}
if info, err := d.Info(); err == nil {
largest = max(largest, info.Size())
}
return nil
})
return float64(largest) / (1 << 20)
}
func tqOpenLogs(root string) float64 {
targets, err := fdTargets()
if err != nil {
return math.NaN()
}
n := 0
for _, target := range targets {
if strings.HasPrefix(target, root) && strings.HasSuffix(target, ".log") {
n++
}
}
return float64(n)
}
func tqFileCount(root string) float64 {
entries, err := os.ReadDir(root)
if err != nil {
return 0
}
return float64(len(entries))
}
func (w *tortureWorld) tqWSURL(nsid string, repo syntax.DID, pipeline string, workflows ...string) string {
q := url.Values{}
q.Set("repo", repo.String())
q.Set("pipeline", pipeline)
for _, name := range workflows {
q.Add("workflows", name)
}
return "ws" + strings.TrimPrefix(w.api.URL, "http") + "/xrpc/" + nsid + "?" + q.Encode()
}
// small recv buffer so a stalled reader backs up the send queue quickly
func tqDialer(smallRecvBuf bool) *websocket.Dialer {
d := &websocket.Dialer{HandshakeTimeout: 30 * time.Second}
if smallRecvBuf {
d.NetDial = func(network, addr string) (net.Conn, error) {
c, err := net.Dial(network, addr)
if err != nil {
return nil, err
}
if tcp, ok := c.(*net.TCPConn); ok {
_ = tcp.SetReadBuffer(4096)
}
return c, nil
}
}
return d
}
func tqDrain(conn *websocket.Conn, bytesRead, msgs *atomic.Int64) {
for {
_, p, err := conn.ReadMessage()
if err != nil {
return
}
bytesRead.Add(int64(len(p)))
msgs.Add(1)
}
}
func init() {
// the only worker gets stuck here and the backlog piles up behind it
registerTorture("queue-backlog", "a big backlog nobody is draining yet, then a restart that works it off", func(r *tortureRun) {
depth := r.load("jobs", 4000)
probes := 50
par := 4
w, repos := tqWorld(r, 10, 3, 1)
if _, err := w.tqEnqueue(repos[0], 1, "hang"); err != nil {
r.t.Fatal(err)
}
if !r.waitFor(30*time.Second, func() bool { return w.eng.running.Load() == 1 }) {
r.t.Fatal("the hung pipeline never started")
}
baseMB := r.gaugeValue("db_mb")
var enqueue latencies
var firstMedian, lastMedian, firstEnqueue time.Duration
var writes latencies
var quiet latencies
for range 50 {
start := time.Now()
if err := w.s.db.StatusPending(models.WorkflowId{PipelineId: "tq-quiet", Name: "ci.yaml"}, nil); err != nil {
r.count("status_write", 1)
}
quiet.since(start)
}
quietWrite := quiet.quantile(0.99)
levels := []int{depth / 8, depth / 4, depth / 2, depth}
for _, level := range levels {
have := int(w.rows("jobs"))
w.tqEnqueueMany(repos, max(level-have, 0), par, 1, &enqueue, "sleep 1ms")
if level == levels[0] {
firstEnqueue = enqueue.quantile(0.99)
}
// probes only pull 50 jobs a level, not enough to shift the depth
var lat latencies
stopWrites := make(chan struct{})
var wg sync.WaitGroup
wg.Go(func() {
wid := models.WorkflowId{PipelineId: "tq-probe", Name: "ci.yaml"}
for {
select {
case <-stopWrites:
return
case <-time.After(2 * time.Millisecond):
}
start := time.Now()
if err := w.s.db.StatusPending(wid, nil); err != nil {
r.count("status_write", 1)
}
writes.since(start)
}
})
for range probes {
start := time.Now()
if _, err := w.s.db.DequeueJob(context.Background(), nil); err != nil {
r.count("dequeue_probe", 1)
}
lat.since(start)
}
close(stopWrites)
wg.Wait()
if level == levels[0] {
firstMedian = lat.quantile(0.5)
}
lastMedian = lat.quantile(0.5)
}
r.latency("enqueue", &enqueue)
r.latency("status_write_during_dequeue", &writes)
have := int(w.rows("jobs"))
r.metric("db_kb_per_queued_job", (r.gaugeValue("db_mb")-baseMB)*1024/float64(max(have, 1)))
r.metric("dequeue_ms_per_1k_queued", float64(lastMedian.Microseconds())/1000/float64(max(have, 1))*1000)
r.check("D1 dequeue cost is flat in queue depth", lastMedian <= 3*firstMedian+500*time.Microsecond,
"median dequeue %v at depth %d vs %v at depth %d", lastMedian, levels[len(levels)-1], firstMedian, levels[0])
r.check("D1 a dequeue doesn't stall other writers", writes.quantile(0.99) <= 10*quietWrite+50*time.Millisecond,
"status write p99 %v while dequeues ran, %v when quiet, want within 10x plus 50ms", writes.quantile(0.99), quietWrite)
r.check("enqueue stays fast under the backlog", enqueue.quantile(0.99) <= 10*firstEnqueue+50*time.Millisecond,
"enqueue p99 %v over the whole run, %v at the first depth, want within 10x plus 50ms", enqueue.quantile(0.99), firstEnqueue)
w.stop()
w.maxJobs = int(r.param("drain_workers", 8))
w.boot()
total := int(w.rows("jobs"))
marks := []float64{0.75, 0.5, 0.25, 0}
at := make([]time.Duration, len(marks))
drainStart := time.Now()
drained := true
for i, m := range marks {
target := int(float64(total) * m)
if !r.waitFor(4*time.Minute, func() bool { return int(w.rows("jobs")) <= target }) {
drained = false
break
}
at[i] = time.Since(drainStart)
}
r.check("queue drains after the restart", drained, "%v jobs still queued", w.rows("jobs"))
if drained && at[0] > 0 && at[3] > at[2] {
first := float64(total) / 4 / at[0].Seconds()
last := float64(total) / 4 / (at[3] - at[2]).Seconds()
r.metric("drain_first_quarter_per_s", first)
r.metric("drain_last_quarter_per_s", last)
r.check("D1 drain rate doesn't depend on how deep the queue is", first >= 0.7*last,
"%.0f jobs/s over the first quarter vs %.0f jobs/s over the last", first, last)
}
r.note("the repo schedule refresh warnings in this run are the records-only repos having no git, harness noise")
r.note("the fake engine does no work, drain rates are about dequeue and bookkeeping only. probe-dequeued jobs and the hung pipeline are left unfinished on purpose")
})
registerTorture("queue-fairness", "one owner floods the queue while a few small owners push a little", func(r *tortureRun) {
flood := r.load("flood", 1500)
smallOwners := 5
smallJobs := 3
workers := int(r.param("workers", 4))
floodStep := r.duration("flood_step", 100*time.Millisecond)
w, repos := tqWorld(r, 1+smallOwners, 1, workers)
floodRepo, smallRepos := repos[0], repos[1:]
floodOwner := floodRepo.owner.String()
w.tqEnqueueMany(repos[:1], flood, 4, 1, nil, "sleep "+floodStep.String())
if !r.waitFor(time.Minute, func() bool { return w.eng.started.Load() >= int64(workers*3) }) {
r.t.Fatal("the flood never started running")
}
ahead := len(w.eng.startOrder())
r.metric("queued_when_smalls_arrive", w.rows("jobs"))
var smallIDs []models.PipelineId
for range smallJobs {
smallIDs = append(smallIDs, w.tqEnqueueMany(smallRepos, smallOwners, 1, 1, nil, "sleep "+floodStep.String())...)
}
done := r.waitFor(5*time.Minute, func() bool {
for _, id := range smallIDs {
if !w.tqWorkflowDone(id, "ci.yaml") {
return false
}
}
return true
})
var delay latencies
w.tqStartDelays(smallIDs, "ci.yaml", &delay)
r.latency("small_owner_start_delay", &delay)
order := w.eng.startOrder()
firstSmall := -1
for i := ahead; i < len(order); i++ {
if order[i] != floodOwner {
firstSmall = i
break
}
}
r.metric("flood_starts_ahead_of_first_small", float64(firstSmall-ahead))
r.metric("flood_starts_total", float64(flood))
r.check("small owners finish while the flood is still queued", done && w.rows("jobs") > 0,
"done=%v queued=%v", done, w.rows("jobs"))
r.check("Q3 a small owner is only ever behind about one round of flood jobs", firstSmall >= 0 && firstSmall-ahead <= 2*workers,
"%d flood jobs started after the small ones were queued and before the first small one", firstSmall-ahead)
r.check("Q3 small owners start within a few job lengths", delay.quantile(0.99) <= 20*floodStep,
"p99 wait %v with %v jobs and %d workers", delay.quantile(0.99), floodStep, workers)
r.check("the flood drains", w.tqFinished(10*time.Minute), "%v workflows unfinished", w.unfinishedWorkflows())
})
registerTorture("queue-restart-mid-run", "restarts with work in flight, and a worker that wakes into a database error", func(r *tortureRun) {
hung := r.load("hung", 6)
queued := r.load("queued", 30)
recoverWait := 20 * time.Second
rounds := r.load("wake_rounds", 20)
lockHold := 7 * time.Second
w, repos := tqWorld(r, 3, 1, hung+2)
startHung := func(tag string) []models.PipelineId {
ids := w.tqEnqueueMany(repos, hung, 2, 1, nil, "hang")
if !r.waitFor(30*time.Second, func() bool { return w.eng.running.Load() >= int64(hung) }) {
r.t.Fatalf("%s: only %d of %d hung workflows started", tag, w.eng.running.Load(), hung)
}
return ids
}
startHung("clean")
w.tqEnqueueMany(repos, queued, 2, 1, nil, "sleep 50ms")
w.stop()
r.check("a clean stop destroys every running workflow", w.eng.running.Load() == 0, "%d still running", w.eng.running.Load())
w.boot()
finished := w.tqFinished(2 * time.Minute)
r.check("queued jobs survive a clean restart and finish", finished, "%v workflows pending or running", w.unfinishedWorkflows())
// erase them here, a killed process never writes its terminal row
crashed := startHung("crash")
w.stop()
w.tqEraseTerminals(crashed)
w.boot()
recovered := r.waitFor(recoverWait, func() bool { return w.unfinishedWorkflows() == 0 })
r.metric("unfinished_after_crash", w.unfinishedWorkflows())
notFailed := 0
for _, id := range crashed {
if id == "" {
continue
}
if status, err := w.s.db.GetStatus(models.WorkflowId{PipelineId: id, Name: "ci.yaml"}); err != nil || status != models.StatusKindFailed {
notFailed++
}
}
r.check("Q1 a crash restart marks what it was running as failed", notFailed == 0, "%d of %d crashed workflows are not failed, want 0", notFailed, len(crashed))
r.check("Q1 nothing stays pending or running forever after a crash restart", recovered,
"%v workflows pending or running %v after the restart, want 0", w.unfinishedWorkflows(), recoverWait)
if !w.tqFinished(time.Minute) {
r.note("workflows from the crash step are still unfinished, wake rounds run on top of them")
}
stuck := 0
for range rounds {
var ids [2]models.PipelineId
var wg sync.WaitGroup
for i := range ids {
wg.Go(func() {
id, err := w.tqEnqueue(repos[i%len(repos)], 1, "hang")
if err != nil {
r.count("enqueue_error", 1)
}
ids[i] = id
})
}
wg.Wait()
if !r.waitFor(10*time.Second, func() bool { return w.eng.running.Load() >= 2 }) {
stuck++
}
for _, id := range ids {
engine.CancelWorkflow(models.WorkflowId{PipelineId: id, Name: "ci.yaml"})
}
r.waitFor(30*time.Second, func() bool { return w.eng.running.Load() == 0 })
}
r.metric("rounds_with_a_second_job_not_started", float64(stuck))
r.check("W2 a second enqueue wakes a second idle worker", stuck == 0,
"%d of %d rounds had a job that never started with idle workers around", stuck, rounds)
r.note("the wake race is narrow, a clean result here doesn't prove the channel is safe")
startedBefore := w.eng.started.Load()
id, tpl, record, err := w.tqPipeline(repos[0], 1, "sleep 10ms")
if err != nil {
r.t.Fatal(err)
}
if err := w.s.db.EnqueuePipeline(context.Background(), repos[0].did.String(), record, nil, tpl, false, "", "", nil); err != nil {
r.t.Fatal(err)
}
conn, err := w.s.db.Conn(context.Background())
if err != nil {
r.t.Fatal(err)
}
if _, err := conn.ExecContext(context.Background(), "begin immediate"); err != nil {
r.t.Fatal(err)
}
select {
case w.s.jobWake <- struct{}{}:
default:
}
time.Sleep(lockHold)
_, _ = conn.ExecContext(context.Background(), "rollback")
conn.Close()
resumed := r.waitFor(15*time.Second, func() bool { return w.eng.started.Load() > startedBefore })
r.check("W2 the queue resumes after a dequeue error with no new enqueue", resumed,
"job %s started=%v within %v of the lock being released, want started", id, resumed, 15*time.Second)
select {
case w.s.jobWake <- struct{}{}:
default:
}
})
registerTorture("logs-flood", "log output without newlines, a lot of output, and many workflows at once", func(r *tortureRun) {
rawMB := 64
dribbleKB := 2048
logMB := 24
logWorkflows := 4
many := r.load("many_workflows", 150)
slots := int(r.param("slots", 4))
capMB := r.param("log_cap_mb", 33)
heapBudget := r.param("heap_budget_mb", 48)
w, repos := tqWorld(r, 1, 1, 4)
eng := w.eng
repo := repos[0]
logDir := filepath.Join(w.dir, "logs")
baseGoroutines := r.gaugeValue("goroutines")
baseFds := r.gaugeValue("fds")
// no newline means the whole thing waits in the writer's pending buffer
rawCmd := fmt.Sprintf("raw %d", rawMB<<20)
// tight gc so heap numbers are live memory, not garbage
defer debug.SetGCPercent(debug.SetGCPercent(10))
runtime.GC()
baseHeap := r.gaugeValue("heap_mb")
peaks := tqPeaks(r, func() {
if _, err := w.tqEnqueue(repo, 1, rawCmd); err != nil {
r.t.Fatal(err)
}
if !w.tqFinished(5 * time.Minute) {
r.check("raw output workflow finishes", false, "still unfinished after 5m")
}
}, "heap_mb", "rss_mb")
debug.SetGCPercent(100)
r.metric("raw_heap_growth_mb", peaks[0]-baseHeap)
r.metric("raw_rss_peak_mb", peaks[1])
r.check("L1 output without newlines doesn't pile up in memory", peaks[0]-baseHeap <= heapBudget,
"heap grew %.0fMB while a step wrote %dMB with no newline, budget %.0fMB", peaks[0]-baseHeap, rawMB, heapBudget)
// doubling the bytes should only double the time, small writes rescan the pending line
dribble := func(kb int) time.Duration {
cmd := fmt.Sprintf("dribble %d", kb<<10)
if _, err := w.tqEnqueue(repo, 1, cmd); err != nil {
r.t.Fatal(err)
}
w.tqFinished(5 * time.Minute)
return eng.tookFor(cmd)
}
once, twice := dribble(dribbleKB), dribble(2*dribbleKB)
r.metric("dribble_step_s", once.Seconds())
r.check("L1 a long line written in small chunks costs linear time", twice <= 3*once+50*time.Millisecond,
"%dKB in 16 byte writes took %v, %dKB took %v, want the double within 3x plus 50ms", dribbleKB, once, 2*dribbleKB, twice)
logCmd := fmt.Sprintf("log %d", logMB<<20)
// sample the size while it still grows, logs get deleted once archived
r.gauge("largest_log_mb", func() float64 { return tqLargestFileMB(logDir) })
largest := tqPeaks(r, func() {
if _, err := w.tqEnqueue(repo, logWorkflows, logCmd); err != nil {
r.t.Fatal(err)
}
w.tqFinished(5 * time.Minute)
}, "largest_log_mb")[0]
r.metric("largest_log_file_mb", largest)
r.check("L1 a workflow's log file is capped", largest > 1 && largest <= capMB,
"largest log file seen while running is %.1fMB, want over 1MB so the flood was seen and at most %.0fMB", largest, capMB)
r.note("log_cap_mb is the default LogMaxBytes (32MB) plus a line and a marker, dawn picks the real number")
w.eng.slots = engine.NewSemaphoreSlotter(slots)
r.gauge("open_workflow_logs", func() float64 { return tqOpenLogs(logDir) })
peaks = tqPeaks(r, func() {
if _, err := w.tqEnqueue(repo, many, "lines 100; sleep 200ms"); err != nil {
r.t.Fatal(err)
}
w.tqFinished(5 * time.Minute)
}, "fds", "goroutines", "open_workflow_logs")
logPeak := peaks[2]
r.metric("many_open_logs_peak", logPeak)
r.check("Q2 log files are only open for workflows that hold a slot", logPeak <= float64(slots)+16,
"%d workflows on %d slots peaked %.0f open workflow log fds, want <= %d (a few more are archives in flight)", many, slots, logPeak, slots+16)
r.note("many_workflows has no cap in spindle (Q2), this only measures what a pipeline that wide costs. the log engine writes real files, the numbers are about spindle's writer and bookkeeping, nothing an engine does")
tqSettleBoth(r, "after the floods", baseGoroutines, baseFds, 64, 30*time.Second)
})
registerTorture("logs-subscribers", "many log subscribers, stalled readers and vanishing clients", func(r *tortureRun) {
subs := r.load("subscribers", 300)
slow := r.load("slow_readers", 60)
churn := r.load("churn", 400)
slowWait := tqLogWriteTimeout + 10*time.Second
backlogMB := 32
w, repos := tqWorld(r, 1, 1, 4)
repo := repos[0]
logDir := filepath.Join(w.dir, "logs")
dbStat := func(read func(sql.DBStats) int) func() float64 {
return func() float64 {
return w.fromSpindle(func(s *Spindle) float64 { return float64(read(s.db.Stats())) })
}
}
r.gauge("db_open_conns", dbStat(func(st sql.DBStats) int { return st.OpenConnections }))
r.gauge("db_conns_in_use", dbStat(func(st sql.DBStats) int { return st.InUse }))
logSize := func(id models.PipelineId) int64 {
fi, err := os.Stat(models.LogFilePath(logDir, models.WorkflowId{PipelineId: id, Name: "ci.yaml"}))
if err != nil {
return 0
}
return fi.Size()
}
small, err := w.tqEnqueue(repo, 1, "lines 200", "hang")
if err != nil {
r.t.Fatal(err)
}
big, err := w.tqEnqueue(repo, 1, fmt.Sprintf("log %d", backlogMB<<20), "hang")
if err != nil {
r.t.Fatal(err)
}
if !r.waitFor(time.Minute, func() bool { return w.eng.running.Load() >= 2 && logSize(big) >= int64(backlogMB)<<20 }) {
r.t.Fatal("the running pipelines never got their logs written")
}
smallURL := w.tqWSURL(org_tangled.CiSubscribePipelineLogsNSID, repo.did, string(small), "ci.yaml")
bigURL := w.tqWSURL(org_tangled.CiSubscribePipelineLogsNSID, repo.did, string(big), "ci.yaml")
// take the baseline after the first viewer warms lazy state
if conn, _, err := tqDialer(false).Dial(smallURL, nil); err == nil {
_ = conn.SetReadDeadline(time.Now().Add(10 * time.Second))
_, _, _ = conn.ReadMessage()
conn.Close()
}
time.Sleep(2 * time.Second)
baseGoroutines := r.gaugeValue("goroutines")
baseFds := r.gaugeValue("fds")
var first latencies
var bytesRead, msgs atomic.Int64
conns := make([]*websocket.Conn, subs)
sem := make(chan struct{}, 50)
var wg sync.WaitGroup
for i := range subs {
wg.Go(func() {
sem <- struct{}{}
start := time.Now()
conn, _, err := tqDialer(false).Dial(smallURL, nil)
<-sem
if err != nil {
r.count("dial", 1)
return
}
_ = conn.SetReadDeadline(time.Now().Add(30 * time.Second))
_, p, err := conn.ReadMessage()
if err != nil {
r.count("first_message", 1)
conn.Close()
return
}
first.since(start)
bytesRead.Add(int64(len(p)))
_ = conn.SetReadDeadline(time.Time{})
conns[i] = conn
go tqDrain(conn, &bytesRead, &msgs)
})
}
wg.Wait()
r.latency("herd_first_message", &first)
r.metric("goroutines_per_subscriber", (r.gaugeValue("goroutines")-baseGoroutines)/float64(subs))
r.metric("fds_per_subscriber", (r.gaugeValue("fds")-baseFds)/float64(subs))
time.Sleep(3 * time.Second)
for _, c := range conns {
if c != nil {
c.Close()
}
}
tqSettleBoth(r, "herd leaves", baseGoroutines, baseFds, 30, 30*time.Second)
// stalled readers, with more backlog than the socket buffers hold
baseGoroutines, baseFds = r.gaugeValue("goroutines"), r.gaugeValue("fds")
var slowConns []*websocket.Conn
for range slow {
conn, _, err := tqDialer(true).Dial(bigURL, nil)
if err != nil {
r.count("slow_dial", 1)
continue
}
slowConns = append(slowConns, conn)
}
time.Sleep(5 * time.Second)
r.metric("slow_goroutines_per_reader_at_5s", (r.gaugeValue("goroutines")-baseGoroutines)/float64(max(len(slowConns), 1)))
time.Sleep(max(slowWait-5*time.Second, 0))
r.metric("slow_fds_over_base", r.gaugeValue("fds")-baseFds)
for _, c := range slowConns {
c.UnderlyingConn().Close()
}
tqSettleBoth(r, "slow readers leave", baseGoroutines, baseFds, 30, 15*time.Second)
// they leave with backlog still unsent on the server
for _, abrupt := range []bool{false, true} {
baseGoroutines, baseFds = r.gaugeValue("goroutines"), r.gaugeValue("fds")
sem = make(chan struct{}, 25)
for range churn {
wg.Go(func() {
sem <- struct{}{}
defer func() { <-sem }()
conn, _, err := tqDialer(false).Dial(smallURL, nil)
if err != nil {
r.count("churn_dial", 1)
return
}
_ = conn.SetReadDeadline(time.Now().Add(10 * time.Second))
_, _, _ = conn.ReadMessage()
if abrupt {
conn.UnderlyingConn().Close()
} else {
conn.Close()
}
})
}
wg.Wait()
phase := "churn with a close frame"
if abrupt {
phase = "churn without a close frame"
}
tqSettleBoth(r, phase, baseGoroutines, baseFds, 30, 15*time.Second)
}
})
registerTorture("db-growth", "pipelines and statuses at a steady rate, what grows and what slows", func(r *tortureRun) {
soak := 3 * time.Minute
perSec := float64(30)
workflows := int(r.param("workflows", 3))
owners := r.load("owners", 10)
w, repos := tqWorld(r, owners, 2, 8)
logDir := filepath.Join(w.dir, "logs")
r.gauge("log_files", func() float64 { return tqFileCount(logDir) })
baseMB := r.gaugeValue("db_mb")
var recentEarly, recentLate, rareEarly, rareLate latencies
probe := func(repo *tortureRepo, elapsed time.Duration) {
query := func(kinds []string, limit int, early, late *latencies) {
start := time.Now()
if _, _, _, err := w.s.db.QueryPipelines(context.Background(), repo.did.String(), nil, "", kinds, limit); err != nil {
r.count("query", 1)
}
switch {
case elapsed < soak/3:
early.since(start)
case elapsed > 2*soak/3:
late.since(start)
}
}
query(nil, 50, &recentEarly, &recentLate)
query([]string{"manual"}, 30, &rareEarly, &rareLate)
}
var enqueue latencies
interval := time.Duration(float64(time.Second) / perSec)
begin := time.Now()
next := begin
lastProbe := begin
sent := 0
for time.Since(begin) < soak {
// half go to repos[0] so its history is long enough for the query probes
repo := repos[0]
if sent%2 == 1 {
repo = repos[sent%len(repos)]
}
if _, err := w.tqEnqueueTimed(&enqueue, repo, workflows, "lines 20"); err != nil {
r.count("enqueue_error", 1)
}
sent++
if time.Since(lastProbe) >= 5*time.Second {
probe(repos[0], time.Since(begin))
lastProbe = time.Now()
}
next = next.Add(interval)
time.Sleep(time.Until(next))
}
r.latency("enqueue", &enqueue)
r.metric("queued_at_end_of_load", w.rows("jobs"))
r.check("the queue keeps up with the load", w.rows("jobs") <= 50,
"%v jobs still queued at the end of the load", w.rows("jobs"))
w.tqFinished(2 * time.Minute)
pipelines := w.rows("pipelines")
statuses := w.rows("workflow_statuses")
growthMB := r.gaugeValue("db_mb") - baseMB
r.metric("pipelines", pipelines)
r.metric("status_rows", statuses)
r.metric("db_kb_per_pipeline", growthMB*1024/max(pipelines, 1))
r.metric("db_gb_per_month_at_this_rate", growthMB*1024/max(pipelines, 1)*perSec*86400*30/1e6)
r.metric("log_dir_files", tqFileCount(logDir))
r.latency("query_recent_early", &recentEarly)
r.latency("query_recent_late", &recentLate)
r.latency("query_rare_kind_early", &rareEarly)
r.latency("query_rare_kind_late", &rareLate)
r.check("QP1 listing a repo's recent pipelines doesn't slow down as history grows",
recentLate.len() == 0 || recentLate.quantile(0.5) <= 2*recentEarly.quantile(0.5)+300*time.Microsecond,
"median %v late vs %v early", recentLate.quantile(0.5), recentEarly.quantile(0.5))
r.check("QP1 a rare kind filter doesn't walk the whole history",
rareLate.len() == 0 || rareLate.quantile(0.5) <= 2*rareEarly.quantile(0.5)+300*time.Microsecond,
"median %v late vs %v early with no matching rows", rareLate.quantile(0.5), rareEarly.quantile(0.5))
archived := w.rows("mill_artifacts")
r.metric("archived_logs", archived)
if archived > 0 {
r.check("G1 a log that is archived leaves local disk", tqFileCount(logDir) <= 10,
"%.0f log files in the log dir with %.0f archived, want <= 10", tqFileCount(logDir), archived)
} else {
r.note("no log was archived in this setup, so log disk growth shows in log_dir_files but isn't checked")
}
r.note("one workflow is %d statuses at least (pending, running, success). the fake engine writes %s of log per workflow, the db numbers are bookkeeping only", 3, "20 lines")
})
}
func (w *tortureWorld) tqEraseTerminals(ids []models.PipelineId) {
d, err := db.Make(context.Background(), w.dbPath())
if err != nil {
w.r.t.Fatal(err)
}
defer d.Close()
for _, id := range ids {
if id == "" {
continue
}
if _, err := d.Exec(`delete from workflow_statuses where pipeline_id = ? and status not in ('pending', 'running')`, string(id)); err != nil {
w.r.t.Fatal(err)
}
}
}